This repository has been archived on 2024-08-21. You can view files and clone it, but cannot push or open issues or pull requests.
TelegramPoster/modules/api_client.py

264 lines
9.5 KiB
Python
Raw Normal View History

2023-02-17 12:51:38 +02:00
"""This is only a temporary solution. Complete Photos API client is yet to be developed."""
2023-02-14 17:25:56 +02:00
import asyncio
from base64 import b64decode, b64encode
2023-01-17 15:11:23 +02:00
from os import makedirs, path, sep
from random import choice
2023-03-12 15:54:06 +02:00
from traceback import print_exc
2023-02-14 17:25:56 +02:00
from typing import Tuple, Union
2023-03-12 15:54:06 +02:00
import aiofiles
2023-03-22 12:02:55 +02:00
from aiohttp import FormData
2023-02-14 17:25:56 +02:00
2023-03-12 23:14:03 +02:00
from classes.exceptions import (
AlbumCreationDuplicateError,
AlbumCreationError,
AlbumCreationNameError,
SubmissionUploadError,
UserCreationDuplicateError,
UserCreationError,
)
2023-02-14 17:25:56 +02:00
from modules.logger import logWrite
2023-03-17 15:44:37 +02:00
from modules.utils import configGet, locale
2023-03-22 12:02:55 +02:00
from modules.http_client import http_session
2023-03-02 17:39:59 +02:00
2023-02-14 17:25:56 +02:00
async def authorize() -> str:
2023-01-17 15:11:23 +02:00
makedirs(configGet("cache", "locations"), exist_ok=True)
2023-03-09 12:33:02 +02:00
if path.exists(configGet("cache", "locations") + sep + "api_access") is True:
async with aiofiles.open(
configGet("cache", "locations") + sep + "api_access", "rb"
) as file:
2023-03-02 23:38:48 +02:00
token = b64decode(await file.read()).decode("utf-8")
2023-03-09 12:33:02 +02:00
if (
await http_session.get(
configGet("address", "posting", "api") + "/users/me/",
headers={"Authorization": f"Bearer {token}"},
)
).status == 200:
2023-01-17 15:11:23 +02:00
return token
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"),
2023-03-09 12:33:02 +02:00
"password": configGet("password", "posting", "api"),
2023-01-17 15:11:23 +02:00
}
2023-03-09 12:33:02 +02:00
response = await http_session.post(
configGet("address", "posting", "api") + "/token", data=payload
)
2023-03-02 23:38:48 +02:00
if not response.ok:
2023-03-09 12:33:02 +02:00
logWrite(
2023-03-17 15:44:37 +02:00
locale(
"api_creds_invalid",
"console",
locale=configGet("locale_log").format(
configGet("address", "posting", "api"),
configGet("username", "posting", "api"),
response.status,
),
)
2023-03-09 12:33:02 +02:00
)
2023-02-14 17:25:56 +02:00
raise ValueError
2023-03-09 12:33:02 +02:00
async with aiofiles.open(
configGet("cache", "locations") + sep + "api_access", "wb"
) as file:
await file.write(
b64encode((await response.json())["access_token"].encode("utf-8"))
)
2023-03-02 17:39:59 +02:00
return (await response.json())["access_token"]
2023-01-17 15:11:23 +02:00
2023-03-09 12:33:02 +02:00
2023-02-14 17:25:56 +02:00
async def random_pic(token: Union[str, None] = None) -> Tuple[str, str]:
2023-01-17 15:11:23 +02:00
"""Returns random image id and filename from the queue.
### Returns:
* `Tuple[str, str]`: First value is an ID and the filename in the filesystem to be indexed.
2023-03-09 12:33:02 +02:00
"""
2023-02-17 12:51:38 +02:00
token = await authorize() if token is None else token
2023-03-09 12:33:02 +02:00
logWrite(
f'{configGet("address", "posting", "api")}/albums/{configGet("album", "posting", "api")}/photos?q=&page_size={configGet("page_size", "posting")}&caption=queue'
)
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,
# )
2023-03-02 17:39:59 +02:00
if resp.status != 200:
2023-03-09 12:33:02 +02:00
logWrite(
2023-03-17 15:44:37 +02:00
locale(
"random_pic_error_code",
"console",
locale=configGet("locale_log").format(
configGet("album", "posting", "api"), resp.status
),
),
2023-03-09 12:33:02 +02:00
)
logWrite(
2023-03-17 15:44:37 +02:00
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,
),
),
2023-03-09 12:33:02 +02:00
debug=True,
)
2023-02-14 17:25:56 +02:00
raise ValueError
2023-03-02 17:39:59 +02:00
if len((await resp.json())["results"]) == 0:
2023-02-14 17:25:56 +02:00
raise KeyError
2023-03-02 17:39:59 +02:00
pic = choice((await resp.json())["results"])
2023-01-17 15:11:23 +02:00
return pic["id"], pic["filename"]
2023-03-09 12:33:02 +02:00
async def upload_pic(
filepath: str, allow_duplicates: bool = False, token: Union[str, None] = None
2023-03-16 16:03:14 +02:00
) -> Tuple[bool, list, Union[str, None]]:
2023-02-17 12:51:38 +02:00
token = await authorize() if token is None else token
2023-01-17 15:11:23 +02:00
try:
pic_name = path.basename(filepath)
2023-03-02 23:38:48 +02:00
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()
2023-03-09 12:33:02 +02:00
formdata.add_field(
"file", file_bytes, filename=pic_name, content_type="image/jpeg"
)
2023-03-02 23:38:48 +02:00
response = await http_session.post(
f'{configGet("address", "posting", "api")}/albums/{configGet("album", "posting", "api")}/photos',
2023-03-09 12:33:02 +02:00
params={
"caption": "queue",
"compress": "false",
"ignore_duplicates": str(allow_duplicates).lower(),
},
2023-03-02 23:38:48 +02:00
headers={"Authorization": f"Bearer {token}"},
2023-03-09 12:33:02 +02:00
data=formdata,
2023-03-02 23:38:48 +02:00
)
2023-03-16 16:03:14 +02:00
response_json = await response.json()
2023-03-02 17:39:59 +02:00
if response.status != 200 and response.status != 409:
2023-03-09 12:33:02 +02:00
logWrite(
2023-03-17 15:44:37 +02:00
locale(
"pic_upload_error",
"console",
locale=configGet("locale_log").format(
filepath, response.status, response.content
),
),
2023-03-09 12:33:02 +02:00
)
raise SubmissionUploadError(
str(filepath), response.status, response.content
)
2023-03-16 16:03:14 +02:00
id = response_json["id"] if "id" in await response.json() else None
2023-01-17 15:11:23 +02:00
duplicates = []
2023-03-16 16:03:14 +02:00
if "duplicates" in response_json:
2023-03-18 22:17:04 +02:00
for index, duplicate in enumerate(response_json["duplicates"]): # type: ignore
2023-03-16 16:03:14 +02:00
if response_json["access_token"] is None:
2023-03-09 12:33:02 +02:00
duplicates.append(
f'`{duplicate["id"]}`:\n{configGet("address_external", "posting", "api")}/photos/{duplicate["id"]}'
)
2023-02-19 21:31:08 +02:00
else:
2023-03-09 12:33:02 +02:00
duplicates.append(
2023-03-16 16:03:14 +02:00
f'`{duplicate["id"]}`:\n{configGet("address_external", "posting", "api")}/token/photo/{response_json["access_token"]}?id={index}'
2023-03-09 12:33:02 +02:00
)
2023-03-16 16:03:14 +02:00
return True, duplicates, id
2023-03-02 23:38:48 +02:00
except Exception as exp:
print_exc()
2023-03-16 16:03:14 +02:00
return False, [], None
2023-03-09 12:33:02 +02:00
async def find_pic(
name: str, caption: Union[str, None] = None, token: Union[str, None] = None
) -> Union[dict, None]:
2023-02-17 12:51:38 +02:00
token = await authorize() if token is None else token
try:
2023-03-09 12:33:02 +02:00
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}"},
)
2023-02-17 12:51:38 +02:00
# logWrite(response.json())
2023-03-02 17:39:59 +02:00
if response.status != 200:
2023-02-17 12:51:38 +02:00
return None
2023-03-02 17:39:59 +02:00
if len((await response.json())["results"]) == 0:
2023-02-17 12:51:38 +02:00
return None
2023-03-02 17:39:59 +02:00
return (await response.json())["results"]
2023-02-17 12:51:38 +02:00
except Exception as exp:
2023-03-09 12:33:02 +02:00
logWrite(
2023-03-17 15:44:37 +02:00
locale(
"find_pic_error",
"console",
locale=configGet("locale_log").format(name, caption, exp),
),
2023-03-09 12:33:02 +02:00
)
2023-02-17 12:51:38 +02:00
return None
2023-01-17 15:11:23 +02:00
2023-03-09 12:33:02 +02:00
2023-02-17 12:51:38 +02:00
async def move_pic(id: str, token: Union[str, None] = None) -> bool:
token = await authorize() if token is None else token
2023-01-17 15:11:23 +02:00
try:
2023-03-21 15:34:25 +02:00
response = await http_session.patch(
2023-03-09 12:33:02 +02:00
f'{configGet("address", "posting", "api")}/photos/{id}?caption=sent',
headers={"Authorization": f"Bearer {token}"},
)
2023-03-21 15:34:25 +02:00
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
2023-01-17 15:11:23 +02:00
return True
except:
return False
2023-03-09 12:33:02 +02:00
2023-03-12 23:14:03 +02:00
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
2023-01-17 15:11:23 +02:00
if __name__ == "__main__":
2023-03-09 12:33:02 +02:00
print(asyncio.run(authorize()))