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
2023-02-19 20:44:00 +01:00

98 lines
5.4 KiB
Python

"""This is only a temporary solution. Complete Photos API client is yet to be developed."""
import asyncio
from base64 import b64decode, b64encode
from os import makedirs, path, sep
from random import choice
from typing import Tuple, Union
from requests import get, patch, post
from classes.exceptions import SubmissionUploadError
from modules.logger import logWrite
from modules.utils import configGet
async def authorize() -> str:
makedirs(configGet("cache", "locations"), exist_ok=True)
if path.exists(configGet("cache", "locations")+sep+"api_access") is True:
with open(configGet("cache", "locations")+sep+"api_access", "rb") as file:
token = b64decode(file.read()).decode("utf-8")
if get(configGet("address", "posting", "api")+"/users/me/", headers={"Authorization": f"Bearer {token}"}).status_code == 200:
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"),
"password": configGet("password", "posting", "api")
}
response = post(configGet("address", "posting", "api")+"/token", data=payload)
if response.status_code != 200:
logWrite(f'Incorrect API credentials! Could not login into "{configGet("address", "posting", "api")}" using login "{configGet("username", "posting", "api")}": HTTP {response.status_code}')
raise ValueError
with open(configGet("cache", "locations")+sep+"api_access", "wb") as file:
file.write(b64encode(response.json()["access_token"].encode("utf-8")))
return 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.
### 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=100&caption=queue')
resp = get(f'{configGet("address", "posting", "api")}/albums/{configGet("album", "posting", "api")}/photos?q=&page_size=100&caption=queue', headers={"Authorization": f"Bearer {token}"})
if resp.status_code != 200:
logWrite(f'Could not get photos from album {configGet("album", "posting", "api")}: HTTP {resp.status_code}')
logWrite(f'Could not get photos from "{configGet("address", "posting", "api")}/albums/{configGet("album", "posting", "api")}/photos?q=&page_size=100&caption=queue" using token "{token}": HTTP {resp.status_code}', debug=True)
raise ValueError
if len(resp.json()["results"]) == 0:
raise KeyError
pic = choice(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]:
token = await authorize() if token is None else token
try:
pic_name = path.basename(filepath)
files = {'file': (pic_name, open(filepath, 'rb'), 'image/jpeg')}
response = post(f'{configGet("address", "posting", "api")}/albums/{configGet("album", "posting", "api")}/photos', params={"caption": "queue", "compress": False, "ignore_duplicates": allow_duplicates}, headers={"Authorization": f"Bearer {token}"}, files=files)
if response.status_code != 200 and response.status_code != 409:
logWrite(f"Could not upload '{filepath}' to API: HTTP {response.status_code} with message '{response.content}'")
raise SubmissionUploadError(str(filepath), response.status_code, response.content)
duplicates = []
if "duplicates" in response.json():
for index, duplicate in enumerate(response.json()["duplicates"]):
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
except:
return False, []
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 = 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_code != 200:
return None
if len(response.json()["results"]) == 0:
return None
return response.json()["results"]
except Exception as exp:
logWrite(f"Could not find image with name '{name}' and caption '{caption}' due to: {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:
patch(f'{configGet("address", "posting", "api")}/photos/{id}?caption=sent', headers={"Authorization": f"Bearer {token}"})
return True
except:
return False
if __name__ == "__main__":
print(asyncio.run(authorize()))