API usage overhaul (#27)
* `/report` command added * Updated to libbot 1.5 * Moved to [PhotosAPI_Client](https://git.end-play.xyz/profitroll/PhotosAPI_Client) v0.5.0 from using self-made API client * Video support (almost stable) * Bug fixes and improvements Co-authored-by: profitroll <vozhd.kk@gmail.com> Reviewed-on: #27
This commit is contained in:
@@ -1,38 +1,87 @@
|
||||
"""This is only a temporary solution. Complete Photos API client is yet to be developed."""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from base64 import b64decode, b64encode
|
||||
from os import makedirs, path, sep
|
||||
from random import choice
|
||||
from traceback import print_exc
|
||||
from typing import Tuple, Union
|
||||
from pathlib import Path
|
||||
from typing import Union
|
||||
|
||||
import aiofiles
|
||||
from aiohttp import FormData
|
||||
|
||||
from classes.exceptions import (
|
||||
AlbumCreationDuplicateError,
|
||||
AlbumCreationError,
|
||||
AlbumCreationNameError,
|
||||
SubmissionUploadError,
|
||||
UserCreationDuplicateError,
|
||||
UserCreationError,
|
||||
from aiohttp import ClientSession
|
||||
from libbot import config_get, i18n, sync
|
||||
from photosapi_client import AuthenticatedClient, Client
|
||||
from photosapi_client.api.default.album_create_albums_post import (
|
||||
asyncio as album_create,
|
||||
)
|
||||
from modules.logger import logWrite
|
||||
from modules.utils import configGet, locale
|
||||
from photosapi_client.api.default.album_delete_album_id_delete import (
|
||||
asyncio as album_delete,
|
||||
)
|
||||
from photosapi_client.api.default.album_find_albums_get import asyncio as album_find
|
||||
from photosapi_client.api.default.login_for_access_token_token_post import sync as login
|
||||
from photosapi_client.api.default.photo_delete_photos_id_delete import (
|
||||
asyncio as photo_delete,
|
||||
)
|
||||
from photosapi_client.api.default.photo_find_albums_album_photos_get import (
|
||||
asyncio as photo_find,
|
||||
)
|
||||
from photosapi_client.api.default.photo_get_photos_id_get import asyncio as photo_get
|
||||
from photosapi_client.api.default.photo_patch_photos_id_patch import (
|
||||
asyncio as photo_patch,
|
||||
)
|
||||
from photosapi_client.api.default.photo_random_albums_album_photos_random_get import (
|
||||
asyncio as photo_random,
|
||||
)
|
||||
from photosapi_client.api.default.photo_upload_albums_album_photos_post import (
|
||||
asyncio_detailed as photo_upload,
|
||||
)
|
||||
from photosapi_client.api.default.user_create_users_post import asyncio as user_create
|
||||
from photosapi_client.api.default.user_me_users_me_get import sync as user_me
|
||||
from photosapi_client.api.default.video_find_albums_album_videos_get import (
|
||||
asyncio as video_find,
|
||||
)
|
||||
from photosapi_client.api.default.video_get_videos_id_get import asyncio as video_get
|
||||
from photosapi_client.api.default.video_patch_videos_id_patch import (
|
||||
asyncio as video_patch,
|
||||
)
|
||||
from photosapi_client.api.default.video_random_albums_album_videos_random_get import (
|
||||
asyncio as video_random,
|
||||
)
|
||||
from photosapi_client.api.default.video_upload_albums_album_videos_post import (
|
||||
asyncio as video_upload,
|
||||
)
|
||||
from photosapi_client.models.body_login_for_access_token_token_post import (
|
||||
BodyLoginForAccessTokenTokenPost,
|
||||
)
|
||||
from photosapi_client.models.body_photo_upload_albums_album_photos_post import (
|
||||
BodyPhotoUploadAlbumsAlbumPhotosPost as BodyPhotoUpload,
|
||||
)
|
||||
from photosapi_client.models.body_video_upload_albums_album_videos_post import (
|
||||
BodyVideoUploadAlbumsAlbumVideosPost as BodyVideoUpload,
|
||||
)
|
||||
from photosapi_client.models.http_validation_error import HTTPValidationError
|
||||
from photosapi_client.models.photo import Photo
|
||||
from photosapi_client.models.photo_search import PhotoSearch
|
||||
from photosapi_client.models.token import Token
|
||||
from photosapi_client.models.video import Video
|
||||
from photosapi_client.models.video_search import VideoSearch
|
||||
from photosapi_client.types import File
|
||||
|
||||
from modules.http_client import http_session
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
async def authorize() -> str:
|
||||
makedirs(configGet("cache", "locations"), exist_ok=True)
|
||||
if path.exists(configGet("cache", "locations") + sep + "api_access") is True:
|
||||
|
||||
async def authorize(custom_session: Union[ClientSession, None] = None) -> str:
|
||||
makedirs(await config_get("cache", "locations"), exist_ok=True)
|
||||
session = http_session if custom_session is None else custom_session
|
||||
|
||||
if path.exists(await config_get("cache", "locations") + sep + "api_access") is True:
|
||||
async with aiofiles.open(
|
||||
configGet("cache", "locations") + sep + "api_access", "rb"
|
||||
await config_get("cache", "locations") + sep + "api_access", "rb"
|
||||
) as file:
|
||||
token = b64decode(await file.read()).decode("utf-8")
|
||||
if (
|
||||
await http_session.get(
|
||||
configGet("address", "posting", "api") + "/users/me/",
|
||||
await session.get(
|
||||
await config_get("address", "posting", "api") + "/users/me/",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
).status == 200:
|
||||
@@ -40,27 +89,28 @@ async def authorize() -> str:
|
||||
payload = {
|
||||
"grant_type": "password",
|
||||
"scope": "me albums.list albums.read albums.write photos.list photos.read photos.write videos.list videos.read videos.write",
|
||||
"username": configGet("username", "posting", "api"),
|
||||
"password": configGet("password", "posting", "api"),
|
||||
"username": await config_get("username", "posting", "api"),
|
||||
"password": await config_get("password", "posting", "api"),
|
||||
}
|
||||
response = await http_session.post(
|
||||
configGet("address", "posting", "api") + "/token", data=payload
|
||||
response = await session.post(
|
||||
await config_get("address", "posting", "api") + "/token", data=payload
|
||||
)
|
||||
if not response.ok:
|
||||
logWrite(
|
||||
locale(
|
||||
logger.warning(
|
||||
i18n._(
|
||||
"api_creds_invalid",
|
||||
"console",
|
||||
locale=configGet("locale_log").format(
|
||||
configGet("address", "posting", "api"),
|
||||
configGet("username", "posting", "api"),
|
||||
locale=(await config_get("locale_log")).format(
|
||||
await config_get("address", "posting", "api"),
|
||||
await config_get("username", "posting", "api"),
|
||||
response.status,
|
||||
),
|
||||
)
|
||||
)
|
||||
raise ValueError
|
||||
async with aiofiles.open(
|
||||
configGet("cache", "locations") + sep + "api_access", "wb"
|
||||
str(Path(f"{await config_get('cache', 'locations')}/api_access")),
|
||||
"wb",
|
||||
) as file:
|
||||
await file.write(
|
||||
b64encode((await response.json())["access_token"].encode("utf-8"))
|
||||
@@ -68,196 +118,36 @@ async def authorize() -> str:
|
||||
return (await response.json())["access_token"]
|
||||
|
||||
|
||||
async def random_pic(token: Union[str, None] = None) -> Tuple[str, str]:
|
||||
"""Returns random image id and filename from the queue.
|
||||
unauthorized_client = Client(
|
||||
base_url=sync.config_get("address", "posting", "api"),
|
||||
timeout=5.0,
|
||||
verify_ssl=True,
|
||||
raise_on_unexpected_status=True,
|
||||
)
|
||||
|
||||
### Returns:
|
||||
* `Tuple[str, str]`: First value is an ID and the filename in the filesystem to be indexed.
|
||||
"""
|
||||
token = await authorize() if token is None else token
|
||||
logWrite(
|
||||
f'{configGet("address", "posting", "api")}/albums/{configGet("album", "posting", "api")}/photos?q=&page_size={configGet("page_size", "posting")}&caption=queue'
|
||||
login_token = login(
|
||||
client=unauthorized_client,
|
||||
form_data=BodyLoginForAccessTokenTokenPost(
|
||||
grant_type="password",
|
||||
scope="me albums.list albums.read albums.write photos.list photos.read photos.write videos.list videos.read videos.write",
|
||||
username=sync.config_get("username", "posting", "api"),
|
||||
password=sync.config_get("password", "posting", "api"),
|
||||
),
|
||||
)
|
||||
|
||||
if not isinstance(login_token, Token):
|
||||
logger.warning(
|
||||
"Could not initialize connection due to invalid token: %s", login_token
|
||||
)
|
||||
resp = await http_session.get(
|
||||
f'{configGet("address", "posting", "api")}/albums/{configGet("album", "posting", "api")}/photos?q=&page_size={configGet("page_size", "posting")}&caption=queue',
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
# logWrite(
|
||||
# locale("random_pic_response", "console", locale=configGet("locale_log")).format(
|
||||
# await resp.json()
|
||||
# ),
|
||||
# debug=True,
|
||||
# )
|
||||
if resp.status != 200:
|
||||
logWrite(
|
||||
locale(
|
||||
"random_pic_error_code",
|
||||
"console",
|
||||
locale=configGet("locale_log").format(
|
||||
configGet("album", "posting", "api"), resp.status
|
||||
),
|
||||
),
|
||||
)
|
||||
logWrite(
|
||||
locale(
|
||||
"random_pic_error_debug",
|
||||
"console",
|
||||
locale=configGet("locale_log").format(
|
||||
configGet("address", "posting", "api"),
|
||||
configGet("album", "posting", "api"),
|
||||
configGet("page_size", "posting"),
|
||||
token,
|
||||
resp.status,
|
||||
),
|
||||
),
|
||||
debug=True,
|
||||
)
|
||||
raise ValueError
|
||||
if len((await resp.json())["results"]) == 0:
|
||||
raise KeyError
|
||||
pic = choice((await resp.json())["results"])
|
||||
return pic["id"], pic["filename"]
|
||||
|
||||
|
||||
async def upload_pic(
|
||||
filepath: str, allow_duplicates: bool = False, token: Union[str, None] = None
|
||||
) -> Tuple[bool, list, Union[str, None]]:
|
||||
token = await authorize() if token is None else token
|
||||
try:
|
||||
pic_name = path.basename(filepath)
|
||||
logWrite(f"Uploading {pic_name} to the API...", debug=True)
|
||||
async with aiofiles.open(filepath, "rb") as f:
|
||||
file_bytes = await f.read()
|
||||
formdata = FormData()
|
||||
formdata.add_field(
|
||||
"file", file_bytes, filename=pic_name, content_type="image/jpeg"
|
||||
)
|
||||
response = await http_session.post(
|
||||
f'{configGet("address", "posting", "api")}/albums/{configGet("album", "posting", "api")}/photos',
|
||||
params={
|
||||
"caption": "queue",
|
||||
"compress": "false",
|
||||
"ignore_duplicates": str(allow_duplicates).lower(),
|
||||
},
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
data=formdata,
|
||||
)
|
||||
response_json = await response.json()
|
||||
if response.status != 200 and response.status != 409:
|
||||
logWrite(
|
||||
locale(
|
||||
"pic_upload_error",
|
||||
"console",
|
||||
locale=configGet("locale_log").format(
|
||||
filepath, response.status, response.content
|
||||
),
|
||||
),
|
||||
)
|
||||
raise SubmissionUploadError(
|
||||
str(filepath), response.status, response.content
|
||||
)
|
||||
id = response_json["id"] if "id" in await response.json() else None
|
||||
duplicates = []
|
||||
if "duplicates" in response_json:
|
||||
for index, duplicate in enumerate(response_json["duplicates"]): # type: ignore
|
||||
if response_json["access_token"] is None:
|
||||
duplicates.append(
|
||||
f'`{duplicate["id"]}`:\n{configGet("address_external", "posting", "api")}/photos/{duplicate["id"]}'
|
||||
)
|
||||
else:
|
||||
duplicates.append(
|
||||
f'`{duplicate["id"]}`:\n{configGet("address_external", "posting", "api")}/token/photo/{response_json["access_token"]}?id={index}'
|
||||
)
|
||||
return True, duplicates, id
|
||||
except Exception as exp:
|
||||
print_exc()
|
||||
return False, [], None
|
||||
|
||||
|
||||
async def find_pic(
|
||||
name: str, caption: Union[str, None] = None, token: Union[str, None] = None
|
||||
) -> Union[dict, None]:
|
||||
token = await authorize() if token is None else token
|
||||
try:
|
||||
response = await http_session.get(
|
||||
f'{configGet("address", "posting", "api")}/albums/{configGet("album", "posting", "api")}/photos',
|
||||
params={"q": name, "caption": caption},
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
# logWrite(response.json())
|
||||
if response.status != 200:
|
||||
return None
|
||||
if len((await response.json())["results"]) == 0:
|
||||
return None
|
||||
return (await response.json())["results"]
|
||||
except Exception as exp:
|
||||
logWrite(
|
||||
locale(
|
||||
"find_pic_error",
|
||||
"console",
|
||||
locale=configGet("locale_log").format(name, caption, exp),
|
||||
),
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
async def move_pic(id: str, token: Union[str, None] = None) -> bool:
|
||||
token = await authorize() if token is None else token
|
||||
try:
|
||||
response = await http_session.patch(
|
||||
f'{configGet("address", "posting", "api")}/photos/{id}?caption=sent',
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
if response.status != 200:
|
||||
logWrite(f"Media moving failed with HTTP {response.status}", debug=True)
|
||||
return False
|
||||
return True
|
||||
except:
|
||||
return False
|
||||
|
||||
|
||||
async def remove_pic(id: str, token: Union[str, None] = None) -> bool:
|
||||
token = await authorize() if token is None else token
|
||||
try:
|
||||
response = await http_session.delete(
|
||||
f'{configGet("address", "posting", "api")}/photos/{id}',
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
if response.status != 204:
|
||||
logWrite(f"Media removal failed with HTTP {response.status}", debug=True)
|
||||
return False
|
||||
return True
|
||||
except:
|
||||
return False
|
||||
|
||||
|
||||
async def create_user(username: str, email: str, password: str) -> None:
|
||||
response = await http_session.post(
|
||||
f'{configGet("address", "posting", "api")}/users',
|
||||
data={"user": username, "email": email, "password": password},
|
||||
)
|
||||
if response.status == 409:
|
||||
raise UserCreationDuplicateError(username)
|
||||
elif response.status != 204:
|
||||
raise UserCreationError(response.status, await response.text(encoding="utf-8"))
|
||||
return None
|
||||
|
||||
|
||||
async def create_album(name: str, title: str) -> None:
|
||||
token = await authorize()
|
||||
response = await http_session.post(
|
||||
f'{configGet("address", "posting", "api")}/albums',
|
||||
params={"name": name, "title": title},
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
if response.status == 409:
|
||||
raise AlbumCreationDuplicateError(name)
|
||||
elif response.status == 406:
|
||||
raise AlbumCreationNameError(await response.json())
|
||||
elif response.status != 200:
|
||||
raise AlbumCreationError(response.status, await response.text(encoding="utf-8"))
|
||||
return None
|
||||
exit()
|
||||
|
||||
client = AuthenticatedClient(
|
||||
base_url=sync.config_get("address", "posting", "api"),
|
||||
timeout=5.0,
|
||||
verify_ssl=True,
|
||||
raise_on_unexpected_status=True,
|
||||
token=login_token.access_token,
|
||||
)
|
||||
|
||||
if __name__ == "__main__":
|
||||
print(asyncio.run(authorize()))
|
||||
|
||||
@@ -1,14 +0,0 @@
|
||||
from modules.utils import configGet
|
||||
from classes.poster_client import PosterClient
|
||||
from convopyro import Conversation
|
||||
|
||||
app = PosterClient(
|
||||
"duptsiaposter",
|
||||
bot_token=configGet("bot_token", "bot"),
|
||||
api_id=configGet("api_id", "bot"),
|
||||
api_hash=configGet("api_hash", "bot"),
|
||||
)
|
||||
|
||||
Conversation(app)
|
||||
|
||||
users_with_context = []
|
||||
@@ -1,78 +0,0 @@
|
||||
import asyncio
|
||||
from sys import exit
|
||||
from traceback import print_exc
|
||||
from modules.api_client import create_album, create_user, http_session
|
||||
from argparse import ArgumentParser
|
||||
|
||||
from modules.utils import configSet
|
||||
|
||||
parser = ArgumentParser(
|
||||
prog="Telegram Poster",
|
||||
description="Bot for posting some of your stuff and also receiving submissions.",
|
||||
)
|
||||
|
||||
parser.add_argument("--create-user", action="store_true")
|
||||
parser.add_argument("--create-album", action="store_true")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
|
||||
async def cli_create_user() -> None:
|
||||
print(
|
||||
"To set up Photos API connection you need to create a new user.\nIf you have email confirmation enabled in your Photos API config - you need to use a real email that will get a confirmation code afterwards.",
|
||||
flush=True,
|
||||
)
|
||||
username = input("Choose username for new Photos API user: ").strip()
|
||||
email = input(f"Choose email for user '{username}': ").strip()
|
||||
password = input(f"Choose password for user '{username}': ").strip()
|
||||
try:
|
||||
result_1 = await create_user(username, email, password)
|
||||
# asyncio.run(create_user(username, email, password))
|
||||
configSet("username", username, "posting", "api")
|
||||
configSet("password", password, "posting", "api")
|
||||
none = input(
|
||||
"Alright. If you have email confirmation enabled - please confirm registration by using the link in your email. After that press Enter. Otherwise just press Enter."
|
||||
)
|
||||
except Exception as exp:
|
||||
print(f"Could not create a user due to {exp}", flush=True)
|
||||
print_exc()
|
||||
exit()
|
||||
if not args.create_album:
|
||||
print("You're done!", flush=True)
|
||||
await http_session.close()
|
||||
exit()
|
||||
return None
|
||||
|
||||
|
||||
async def cli_create_album() -> None:
|
||||
print(
|
||||
"To use Photos API your user needs to have an album to store its data.\nThis wizard will help you to create a new album with its name and title.",
|
||||
flush=True,
|
||||
)
|
||||
name = input("Choose a name for your album: ").strip()
|
||||
title = input(f"Choose a title for album '{name}': ").strip()
|
||||
try:
|
||||
result_2 = await create_album(name, title)
|
||||
# asyncio.run(create_album(name, title))
|
||||
configSet("album", name, "posting", "api")
|
||||
except Exception as exp:
|
||||
print(f"Could not create an album due to {exp}", flush=True)
|
||||
print_exc()
|
||||
exit()
|
||||
print("You're done!", flush=True)
|
||||
await http_session.close()
|
||||
exit()
|
||||
return None
|
||||
|
||||
|
||||
if args.create_user or args.create_album:
|
||||
loop = asyncio.get_event_loop()
|
||||
tasks = []
|
||||
|
||||
if args.create_user:
|
||||
loop.run_until_complete(asyncio.wait([loop.create_task(cli_create_user())]))
|
||||
|
||||
if args.create_album:
|
||||
loop.run_until_complete(asyncio.wait([loop.create_task(cli_create_album())]))
|
||||
|
||||
loop.close()
|
||||
@@ -1,57 +0,0 @@
|
||||
from os import listdir
|
||||
from classes.poster_client import PosterClient
|
||||
from pyrogram.types import BotCommand, BotCommandScopeChat
|
||||
from modules.utils import configGet, locale
|
||||
|
||||
|
||||
async def register_commands(app: PosterClient) -> None:
|
||||
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", "")
|
||||
),
|
||||
)
|
||||
)
|
||||
await app.set_bot_commands(
|
||||
commands_list, language_code=entry.replace(".json", "")
|
||||
)
|
||||
|
||||
# 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")),
|
||||
)
|
||||
)
|
||||
await app.set_bot_commands(commands_list)
|
||||
|
||||
# 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"))
|
||||
)
|
||||
)
|
||||
|
||||
for admin in app.admins:
|
||||
await app.set_bot_commands(
|
||||
commands_admin_list, scope=BotCommandScopeChat(chat_id=admin)
|
||||
)
|
||||
@@ -1,67 +0,0 @@
|
||||
from datetime import datetime
|
||||
from gzip import open as gzipopen
|
||||
from os import getcwd, makedirs, path, stat
|
||||
from shutil import copyfileobj
|
||||
|
||||
from ujson import loads
|
||||
|
||||
with open(getcwd() + path.sep + "config.json", "r", encoding="utf8") as file:
|
||||
json_contents = loads(file.read())
|
||||
log_size = json_contents["logging"]["size"]
|
||||
log_folder = json_contents["logging"]["location"]
|
||||
file.close()
|
||||
|
||||
|
||||
# Check latest log size
|
||||
def checkSize(debug=False) -> None:
|
||||
global log_folder
|
||||
|
||||
if debug:
|
||||
log_file = "debug.log"
|
||||
else:
|
||||
log_file = "latest.log"
|
||||
|
||||
try:
|
||||
makedirs(log_folder, exist_ok=True)
|
||||
log = stat(path.join(log_folder, log_file))
|
||||
if (log.st_size / 1024) > log_size:
|
||||
with open(path.join(log_folder, log_file), "rb") as f_in:
|
||||
with gzipopen(
|
||||
path.join(
|
||||
log_folder,
|
||||
f'{datetime.now().strftime("%d.%m.%Y_%H:%M:%S")}.log.gz',
|
||||
),
|
||||
"wb",
|
||||
) as f_out:
|
||||
copyfileobj(f_in, f_out)
|
||||
print(
|
||||
f'Copied {path.join(log_folder, datetime.now().strftime("%d.%m.%Y_%H:%M:%S"))}.log.gz'
|
||||
)
|
||||
open(path.join(log_folder, log_file), "w").close()
|
||||
except FileNotFoundError:
|
||||
print(f"Log file {path.join(log_folder, log_file)} does not exist")
|
||||
pass
|
||||
|
||||
|
||||
# Append string to log
|
||||
def logAppend(message, debug=False) -> None:
|
||||
global log_folder
|
||||
|
||||
message_formatted = f'[{datetime.now().strftime("%d.%m.%Y")}] [{datetime.now().strftime("%H:%M:%S")}] {message}'
|
||||
checkSize(debug=debug)
|
||||
|
||||
if debug:
|
||||
log_file = "debug.log"
|
||||
else:
|
||||
log_file = "latest.log"
|
||||
|
||||
log = open(path.join(log_folder, log_file), "a")
|
||||
log.write(f"{message_formatted}\n")
|
||||
log.close()
|
||||
|
||||
|
||||
# Print to stdout and then to log
|
||||
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)
|
||||
@@ -1,31 +1,3 @@
|
||||
from datetime import datetime, timedelta
|
||||
from apscheduler.schedulers.asyncio import AsyncIOScheduler
|
||||
from pytimeparse.timeparse import timeparse
|
||||
from modules.utils import configGet
|
||||
from modules.sender import send_content
|
||||
from modules.commands_register import register_commands
|
||||
from modules.app import app
|
||||
|
||||
scheduler = AsyncIOScheduler()
|
||||
|
||||
if configGet("post", "mode"):
|
||||
if configGet("use_interval", "posting"):
|
||||
scheduler.add_job(
|
||||
send_content,
|
||||
"interval",
|
||||
seconds=timeparse(configGet("interval", "posting")),
|
||||
args=[app],
|
||||
)
|
||||
else:
|
||||
for entry in configGet("time", "posting"):
|
||||
dt_obj = datetime.strptime(entry, "%H:%M")
|
||||
scheduler.add_job(
|
||||
send_content, "cron", hour=dt_obj.hour, minute=dt_obj.minute, args=[app]
|
||||
)
|
||||
|
||||
scheduler.add_job(
|
||||
register_commands,
|
||||
"date",
|
||||
run_date=datetime.now() + timedelta(seconds=10),
|
||||
args=[app],
|
||||
)
|
||||
|
||||
@@ -1,113 +1,183 @@
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from os import makedirs, path
|
||||
from random import choice
|
||||
from random import choice, sample
|
||||
from shutil import rmtree
|
||||
from traceback import format_exc
|
||||
from traceback import format_exc, print_exc
|
||||
from typing import Union
|
||||
from uuid import uuid4
|
||||
from PIL import Image
|
||||
|
||||
import aiofiles
|
||||
from aiohttp import ClientSession
|
||||
from libbot.pyrogram.classes import PyroClient
|
||||
from photosapi_client.errors import UnexpectedStatus
|
||||
from PIL import Image
|
||||
|
||||
from classes.poster_client import PosterClient
|
||||
|
||||
from modules.api_client import authorize, move_pic, random_pic, http_session
|
||||
from modules.api_client import (
|
||||
File,
|
||||
PhotoSearch,
|
||||
VideoSearch,
|
||||
authorize,
|
||||
client,
|
||||
photo_get,
|
||||
photo_patch,
|
||||
photo_random,
|
||||
video_get,
|
||||
video_patch,
|
||||
video_random,
|
||||
)
|
||||
from modules.database import col_sent, col_submitted
|
||||
from modules.logger import logWrite
|
||||
from modules.utils import configGet, locale
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def send_content(app: PosterClient) -> None:
|
||||
async def send_content(app: PyroClient, http_session: ClientSession) -> None:
|
||||
try:
|
||||
try:
|
||||
token = await authorize()
|
||||
token = await authorize(http_session)
|
||||
except ValueError:
|
||||
await app.send_message(
|
||||
app.owner,
|
||||
locale("api_creds_invalid", "message", locale=configGet("locale")),
|
||||
app._("api_creds_invalid", "message"),
|
||||
)
|
||||
return
|
||||
|
||||
try:
|
||||
pic = await random_pic()
|
||||
except KeyError:
|
||||
logWrite(locale("post_empty", "console", locale=configGet("locale")))
|
||||
if configGet("error", "reports"):
|
||||
await app.send_message(
|
||||
app.owner,
|
||||
locale("api_queue_empty", "message", locale=configGet("locale")),
|
||||
funcs = []
|
||||
|
||||
if app.config["posting"]["types"]["photo"]:
|
||||
funcs.append((photo_random, photo_get, app.send_photo, photo_patch))
|
||||
|
||||
if app.config["posting"]["types"]["video"]:
|
||||
funcs.append((video_random, video_get, app.send_video, video_patch))
|
||||
|
||||
if not funcs:
|
||||
raise KeyError(
|
||||
"No media source provided: all seem to be disabled in config"
|
||||
)
|
||||
return
|
||||
except ValueError:
|
||||
if configGet("error", "reports"):
|
||||
|
||||
if len(funcs) > 1:
|
||||
found = False
|
||||
for func_iter in sample(funcs, len(funcs)):
|
||||
func = func_iter
|
||||
|
||||
random_results = (
|
||||
await func_iter[0](
|
||||
album=app.config["posting"]["api"]["album"],
|
||||
caption="queue",
|
||||
client=client,
|
||||
limit=1,
|
||||
)
|
||||
).results
|
||||
|
||||
if not random_results:
|
||||
continue
|
||||
|
||||
media: Union[PhotoSearch, VideoSearch] = random_results[0]
|
||||
|
||||
try:
|
||||
response: File = await func_iter[1](id=media.id, client=client)
|
||||
except Exception as exp:
|
||||
print_exc()
|
||||
logger.error("Media is invalid: %s", exp)
|
||||
if app.config["reports"]["error"]:
|
||||
await app.send_message(
|
||||
app.owner, f"Media is invalid: {exp}"
|
||||
)
|
||||
return
|
||||
|
||||
found = True
|
||||
break
|
||||
|
||||
if not found:
|
||||
raise KeyError("No media found")
|
||||
else:
|
||||
func = funcs[0]
|
||||
media: Union[PhotoSearch, VideoSearch] = (
|
||||
await func[0](
|
||||
album=app.config["posting"]["api"]["album"],
|
||||
caption="queue",
|
||||
client=client,
|
||||
limit=1,
|
||||
)
|
||||
).results[0]
|
||||
try:
|
||||
response: File = await func[1](id=media.id, client=client)
|
||||
except Exception as exp:
|
||||
print_exc()
|
||||
logger.error("Media is invalid: %s", exp)
|
||||
if app.config["reports"]["error"]:
|
||||
await app.send_message(app.owner, f"Media is invalid: {exp}")
|
||||
return
|
||||
|
||||
except (KeyError, AttributeError, TypeError, IndexError):
|
||||
logger.info(app._("post_empty", "console"))
|
||||
if app.config["reports"]["error"]:
|
||||
await app.send_message(
|
||||
app.owner,
|
||||
locale("api_queue_error", "message", locale=configGet("locale")),
|
||||
app._("api_queue_empty", "message"),
|
||||
)
|
||||
return
|
||||
|
||||
response = await http_session.get(
|
||||
f'{configGet("address", "posting", "api")}/photos/{pic[0]}',
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
|
||||
if response.status != 200:
|
||||
logWrite(
|
||||
locale(
|
||||
"post_invalid_pic", "console", locale=configGet("locale")
|
||||
).format(response.status, str(await response.json()))
|
||||
)
|
||||
if configGet("error", "reports"):
|
||||
except (ValueError, UnexpectedStatus):
|
||||
if app.config["reports"]["error"]:
|
||||
await app.send_message(
|
||||
app.owner,
|
||||
locale(
|
||||
"post_invalid_pic", "message", locale=configGet("locale")
|
||||
).format(response.status, await response.json()),
|
||||
app._("api_queue_error", "message"),
|
||||
)
|
||||
return
|
||||
|
||||
tmp_dir = str(uuid4())
|
||||
|
||||
makedirs(path.join(configGet("tmp", "locations"), tmp_dir), exist_ok=True)
|
||||
makedirs(path.join(app.config["locations"]["tmp"], tmp_dir), exist_ok=True)
|
||||
|
||||
tmp_path = path.join(tmp_dir, pic[1])
|
||||
tmp_path = path.join(tmp_dir, media.filename)
|
||||
|
||||
async with aiofiles.open(
|
||||
path.join(configGet("tmp", "locations"), tmp_path), "wb"
|
||||
path.join(app.config["locations"]["tmp"], tmp_path), "wb"
|
||||
) as out_file:
|
||||
await out_file.write(await response.read())
|
||||
await out_file.write(response.payload.read())
|
||||
|
||||
logWrite(
|
||||
f'Candidate {pic[1]} ({pic[0]}) is {path.getsize(path.join(configGet("tmp", "locations"), tmp_path))} bytes big',
|
||||
debug=True,
|
||||
logger.info(
|
||||
"Candidate %s (%s) is %s bytes big",
|
||||
media.filename,
|
||||
media.id,
|
||||
path.getsize(path.join(app.config["locations"]["tmp"], tmp_path)),
|
||||
)
|
||||
|
||||
if path.getsize(path.join(configGet("tmp", "locations"), tmp_path)) > 5242880:
|
||||
image = Image.open(path.join(configGet("tmp", "locations"), tmp_path))
|
||||
if (
|
||||
path.getsize(path.join(app.config["locations"]["tmp"], tmp_path)) > 5242880
|
||||
) and func[0] is photo_random:
|
||||
image = Image.open(path.join(app.config["locations"]["tmp"], tmp_path))
|
||||
width, height = image.size
|
||||
image = image.resize((int(width / 2), int(height / 2)), Image.ANTIALIAS)
|
||||
if tmp_path.lower().endswith(".jpeg") or tmp_path.lower().endswith(".jpg"):
|
||||
image.save(
|
||||
path.join(configGet("tmp", "locations"), tmp_path),
|
||||
path.join(app.config["locations"]["tmp"], tmp_path),
|
||||
"JPEG",
|
||||
optimize=True,
|
||||
quality=50,
|
||||
)
|
||||
elif tmp_path.lower().endswith(".png"):
|
||||
image.save(
|
||||
path.join(configGet("tmp", "locations"), tmp_path),
|
||||
path.join(app.config["locations"]["tmp"], tmp_path),
|
||||
"PNG",
|
||||
optimize=True,
|
||||
compress_level=8,
|
||||
)
|
||||
image.close()
|
||||
|
||||
if path.getsize(path.join(configGet("tmp", "locations"), tmp_path)) > 5242880:
|
||||
if (
|
||||
path.getsize(path.join(app.config["locations"]["tmp"], tmp_path)) > 5242880
|
||||
) and func[0] is photo_random:
|
||||
rmtree(
|
||||
path.join(configGet("tmp", "locations"), tmp_dir), ignore_errors=True
|
||||
path.join(app.config["locations"]["tmp"], tmp_dir), ignore_errors=True
|
||||
)
|
||||
raise BytesWarning
|
||||
|
||||
del response
|
||||
|
||||
submitted = col_submitted.find_one({"temp.file": pic[1]})
|
||||
submitted = col_submitted.find_one({"temp.file": media.filename})
|
||||
|
||||
if submitted is not None and submitted["caption"] is not None:
|
||||
caption = submitted["caption"].strip()
|
||||
@@ -116,86 +186,82 @@ async def send_content(app: PosterClient) -> None:
|
||||
|
||||
if (
|
||||
submitted is not None
|
||||
and configGet("enabled", "posting", "submitted_caption")
|
||||
and app.config["posting"]["submitted_caption"]["enabled"]
|
||||
and (
|
||||
(submitted["user"] not in app.admins)
|
||||
or (configGet("ignore_admins", "posting", "submitted_caption") is False)
|
||||
or (
|
||||
app.config["posting"]["submitted_caption"]["ignore_admins"] is False
|
||||
)
|
||||
)
|
||||
):
|
||||
caption = (
|
||||
f"{caption}\n\n{configGet('text', 'posting', 'submitted_caption')}\n"
|
||||
f"{caption}\n\n{app.config['posting']['submitted_caption']['text']}\n"
|
||||
)
|
||||
else:
|
||||
caption = f"{caption}\n\n"
|
||||
|
||||
if configGet("enabled", "caption"):
|
||||
if configGet("link", "caption") != None:
|
||||
caption = f"{caption}[{choice(configGet('text', 'caption'))}]({configGet('link', 'caption')})"
|
||||
if app.config["caption"]["enabled"]:
|
||||
if app.config["caption"]["link"] is not None:
|
||||
caption = f"{caption}[{choice(app.config['caption']['text'])}]({app.config['caption']['link']})"
|
||||
else:
|
||||
caption = f"{caption}{choice(configGet('text', 'caption'))}"
|
||||
caption = f"{caption}{choice(app.config['caption']['text'])}"
|
||||
else:
|
||||
caption = caption
|
||||
|
||||
try:
|
||||
sent = await app.send_photo(
|
||||
configGet("channel", "posting"),
|
||||
path.join(configGet("tmp", "locations"), tmp_path),
|
||||
sent = await func[2](
|
||||
app.config["posting"]["channel"],
|
||||
path.join(app.config["locations"]["tmp"], tmp_path),
|
||||
caption=caption,
|
||||
disable_notification=configGet("silent", "posting"),
|
||||
disable_notification=app.config["posting"]["silent"],
|
||||
)
|
||||
except Exception as exp:
|
||||
logWrite(f"Could not send image {pic[1]} ({pic[0]}) due to {exp}")
|
||||
if configGet("error", "reports"):
|
||||
logger.error(
|
||||
"Could not send media %s (%s) due to %s", media.filename, media.id, exp
|
||||
)
|
||||
if app.config["reports"]["error"]:
|
||||
await app.send_message(
|
||||
app.owner,
|
||||
locale(
|
||||
"post_exception", "message", locale=configGet("locale")
|
||||
).format(exp, format_exc()),
|
||||
app._("post_exception", "message").format(exp, format_exc()),
|
||||
)
|
||||
# rmtree(path.join(configGet("tmp", "locations"), tmp_dir), ignore_errors=True)
|
||||
# rmtree(path.join(app.config['locations']['tmp'], tmp_dir), ignore_errors=True)
|
||||
return
|
||||
|
||||
col_sent.insert_one(
|
||||
{
|
||||
"date": datetime.now(),
|
||||
"image": pic[0],
|
||||
"filename": pic[1],
|
||||
"channel": configGet("channel", "posting"),
|
||||
"image": media.id,
|
||||
"filename": media.filename,
|
||||
"channel": app.config["posting"]["channel"],
|
||||
"caption": None
|
||||
if (submitted is None or submitted["caption"] is None)
|
||||
else submitted["caption"].strip(),
|
||||
}
|
||||
)
|
||||
|
||||
await move_pic(pic[0])
|
||||
await func[3](id=media.id, client=client, caption="sent")
|
||||
|
||||
rmtree(path.join(configGet("tmp", "locations"), tmp_dir), ignore_errors=True)
|
||||
rmtree(path.join(app.config["locations"]["tmp"], tmp_dir), ignore_errors=True)
|
||||
|
||||
logWrite(
|
||||
locale("post_sent", "console", locale=configGet("locale")).format(
|
||||
pic[0],
|
||||
str(configGet("channel", "posting")),
|
||||
logger.info(
|
||||
app._("post_sent", "console").format(
|
||||
media.id,
|
||||
str(app.config["posting"]["channel"]),
|
||||
caption.replace("\n", "%n"),
|
||||
str(configGet("silent", "posting")),
|
||||
str(app.config["posting"]["silent"]),
|
||||
)
|
||||
)
|
||||
|
||||
except Exception as exp:
|
||||
logWrite(
|
||||
locale("post_exception", "console", locale=configGet("locale")).format(
|
||||
str(exp), format_exc()
|
||||
)
|
||||
)
|
||||
if configGet("error", "reports"):
|
||||
logger.error(app._("post_exception", "console").format(str(exp), format_exc()))
|
||||
if app.config["reports"]["error"]:
|
||||
await app.send_message(
|
||||
app.owner,
|
||||
locale("post_exception", "message", locale=configGet("locale")).format(
|
||||
exp, format_exc()
|
||||
),
|
||||
app._("post_exception", "message").format(exp, format_exc()),
|
||||
)
|
||||
try:
|
||||
rmtree(
|
||||
path.join(configGet("tmp", "locations"), tmp_dir), ignore_errors=True
|
||||
path.join(app.config["locations"]["tmp"], tmp_dir), ignore_errors=True
|
||||
)
|
||||
except:
|
||||
pass
|
||||
|
||||
245
modules/utils.py
245
modules/utils.py
@@ -1,252 +1,33 @@
|
||||
from os import kill, makedirs
|
||||
from os import name as osname
|
||||
from os import path, sep
|
||||
from sys import exit
|
||||
from traceback import print_exc
|
||||
from typing import Any
|
||||
import logging
|
||||
from os import makedirs, path
|
||||
from pathlib import Path
|
||||
from typing import List, Union
|
||||
from zipfile import ZipFile
|
||||
|
||||
import aiofiles
|
||||
from ujson import JSONDecodeError, dumps, loads
|
||||
|
||||
from modules.logger import logWrite
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
default_config = {
|
||||
"locale": "en",
|
||||
"locale_log": "en",
|
||||
"locale_fallback": "en",
|
||||
"owner": 0,
|
||||
"admins": [],
|
||||
"bot": {"api_id": 0, "api_hash": "", "bot_token": ""},
|
||||
"database": {
|
||||
"user": None,
|
||||
"password": None,
|
||||
"host": "127.0.0.1",
|
||||
"port": 27017,
|
||||
"name": "tgposter",
|
||||
},
|
||||
"mode": {"post": True, "submit": True},
|
||||
"reports": {"sent": False, "error": True, "startup": True, "shutdown": True},
|
||||
"logging": {"size": 512, "location": "logs"},
|
||||
"locations": {
|
||||
"tmp": "tmp",
|
||||
"data": "data",
|
||||
"cache": "cache",
|
||||
"sent": "data/sent",
|
||||
"queue": "data/queue",
|
||||
"index": "data/index.json",
|
||||
"locale": "locale",
|
||||
},
|
||||
"posting": {
|
||||
"channel": 0,
|
||||
"silent": False,
|
||||
"move_sent": False,
|
||||
"use_interval": False,
|
||||
"interval": "1h30m",
|
||||
"page_size": 300,
|
||||
"submitted_caption": {
|
||||
"enabled": True,
|
||||
"ignore_admins": True,
|
||||
"text": "#submitted",
|
||||
},
|
||||
"extensions": {
|
||||
"photo": ["jpg", "png", "gif", "jpeg"],
|
||||
"video": ["mp4", "avi", "mkv", "webm", "mov"],
|
||||
},
|
||||
"time": [
|
||||
"08:00",
|
||||
"10:00",
|
||||
"12:00",
|
||||
"14:00",
|
||||
"16:00",
|
||||
"18:00",
|
||||
"20:00",
|
||||
"22:00",
|
||||
],
|
||||
"api": {
|
||||
"address": "http://localhost:8054",
|
||||
"address_external": "https://photos.domain.com",
|
||||
"username": "",
|
||||
"password": "",
|
||||
"album": "",
|
||||
},
|
||||
},
|
||||
"caption": {"enabled": False, "link": None, "text": ["sample text"]},
|
||||
"submission": {
|
||||
"timeout": 30,
|
||||
"file_size": 15728640,
|
||||
"tmp_size": 15728640,
|
||||
"allow_duplicates": False,
|
||||
"send_uploaded_id": False,
|
||||
"require_confirmation": {"users": True, "admins": True},
|
||||
"mime_types": [
|
||||
"image/png",
|
||||
"image/gif",
|
||||
"image/jpeg",
|
||||
"video/mp4",
|
||||
"video/quicktime",
|
||||
],
|
||||
},
|
||||
"commands": ["start", "rules"],
|
||||
"commands_admin": ["import", "export", "shutdown"],
|
||||
}
|
||||
USERS_WITH_CONTEXT: List[int] = []
|
||||
|
||||
|
||||
def jsonLoad(filename: str) -> Any:
|
||||
"""Loads arg1 as json and returns its contents"""
|
||||
with open(filename, "r", encoding="utf8") as file:
|
||||
try:
|
||||
output = loads(file.read())
|
||||
except JSONDecodeError:
|
||||
logWrite(
|
||||
f"Could not load json file {filename}: file seems to be incorrect!\n{print_exc()}"
|
||||
)
|
||||
raise
|
||||
except FileNotFoundError:
|
||||
logWrite(
|
||||
f"Could not load json file {filename}: file does not seem to exist!\n{print_exc()}"
|
||||
)
|
||||
raise
|
||||
file.close()
|
||||
return output
|
||||
|
||||
|
||||
def jsonSave(contents: Any, filename: str) -> None:
|
||||
"""Dumps dict/list arg1 to file arg2"""
|
||||
try:
|
||||
with open(filename, "w", encoding="utf8") as file:
|
||||
file.write(
|
||||
dumps(
|
||||
contents, ensure_ascii=False, indent=4, escape_forward_slashes=False
|
||||
)
|
||||
)
|
||||
file.close()
|
||||
except Exception as exp:
|
||||
logWrite(f"Could not save json file {filename}: {exp}\n{print_exc()}")
|
||||
return
|
||||
|
||||
|
||||
def configSet(key: str, value, *args: str):
|
||||
"""Set key to a value
|
||||
Args:
|
||||
* key (str): The last key of the keys path.
|
||||
* value (str/int/float/list/dict/None): Some needed value.
|
||||
* *args (str): Path to key like: dict[args][key].
|
||||
"""
|
||||
this_dict = jsonLoad("config.json")
|
||||
string = "this_dict"
|
||||
for arg in args:
|
||||
string += f'["{arg}"]'
|
||||
if type(value) in [str]:
|
||||
string += f'["{key}"] = "{value}"'
|
||||
else:
|
||||
string += f'["{key}"] = {value}'
|
||||
exec(string)
|
||||
jsonSave(this_dict, "config.json")
|
||||
return
|
||||
|
||||
|
||||
def configGet(key: str, *args: str):
|
||||
"""Get value of the config key
|
||||
Args:
|
||||
* key (str): The last key of the keys path.
|
||||
* *args (str): Path to key like: dict[args][key].
|
||||
Returns:
|
||||
* any: Value of provided key
|
||||
"""
|
||||
this_dict = jsonLoad("config.json")
|
||||
try:
|
||||
this_key = this_dict
|
||||
for dict_key in args:
|
||||
this_key = this_key[dict_key]
|
||||
this_key[key]
|
||||
except KeyError:
|
||||
print(
|
||||
f"Could not find config key '{key}' under path {args}: falling back to default config",
|
||||
flush=True,
|
||||
)
|
||||
this_key = default_config
|
||||
for dict_key in args:
|
||||
this_key = this_key[dict_key]
|
||||
configSet(key, this_key[key], *args)
|
||||
return this_key[key]
|
||||
|
||||
|
||||
def locale(key: str, *args: str, locale=configGet("locale")):
|
||||
"""Get value of locale string
|
||||
Args:
|
||||
* key (str): The last key of the locale's keys path.
|
||||
* *args (list): Path to key like: dict[args][key].
|
||||
* locale (str): Locale to looked up in. Defaults to config's locale value.
|
||||
Returns:
|
||||
* any: Value of provided locale key
|
||||
"""
|
||||
if locale == None:
|
||||
locale = configGet("locale")
|
||||
|
||||
try:
|
||||
this_dict = jsonLoad(f'{configGet("locale", "locations")}{sep}{locale}.json')
|
||||
except FileNotFoundError:
|
||||
try:
|
||||
this_dict = jsonLoad(
|
||||
f'{configGet("locale", "locations")}{sep}{configGet("locale")}.json'
|
||||
)
|
||||
except FileNotFoundError:
|
||||
try:
|
||||
this_dict = jsonLoad(
|
||||
f'{configGet("locale_fallback", "locations")}{sep}{configGet("locale")}.json'
|
||||
)
|
||||
except:
|
||||
return f'⚠️ Locale in config is invalid: could not get "{key}" in {str(args)} from locale "{locale}"'
|
||||
|
||||
this_key = this_dict
|
||||
for dict_key in args:
|
||||
this_key = this_key[dict_key]
|
||||
|
||||
try:
|
||||
return this_key[key]
|
||||
except KeyError:
|
||||
return f'⚠️ Locale in config is invalid: could not get "{key}" in {str(args)} from locale "{locale}"'
|
||||
|
||||
|
||||
async def extract_and_save(handle: ZipFile, filename: str, destpath: str):
|
||||
async def extract_and_save(handle: ZipFile, filename: str, destpath: Union[str, Path]):
|
||||
"""Extract and save file from archive
|
||||
|
||||
Args:
|
||||
* handle (ZipFile): ZipFile handler
|
||||
* filename (str): File base name
|
||||
* path (str): Path where to store
|
||||
### Args:
|
||||
* handle (`ZipFile`): ZipFile handler
|
||||
* filename (`str`): File base name
|
||||
* path (`Union[str, Path]`): Path where to store
|
||||
"""
|
||||
data = handle.read(filename)
|
||||
filepath = path.join(destpath, filename)
|
||||
filepath = path.join(str(destpath), filename)
|
||||
try:
|
||||
makedirs(path.dirname(filepath), exist_ok=True)
|
||||
async with aiofiles.open(filepath, "wb") as fd:
|
||||
await fd.write(data)
|
||||
logWrite(f"Unzipped {filename}", debug=True)
|
||||
logger.debug("Unzipped %s", filename)
|
||||
except IsADirectoryError:
|
||||
makedirs(filepath, exist_ok=True)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
return
|
||||
|
||||
|
||||
try:
|
||||
from psutil import Process
|
||||
except ModuleNotFoundError:
|
||||
print(locale("deps_missing", "console", locale=configGet("locale")), flush=True)
|
||||
exit()
|
||||
|
||||
|
||||
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:
|
||||
Process(pid).kill()
|
||||
|
||||
Reference in New Issue
Block a user