"""This is only a temporary solution. Complete Photos API client is yet to be developed.""" import asyncio from photosapi_client import AuthenticatedClient, Client from photosapi_client.api.default.user_me_users_me_get import sync as user_me from photosapi_client.api.default.user_create_users_post import asyncio as user_create from photosapi_client.api.default.login_for_access_token_token_post import ( sync as login, ) from photosapi_client.api.default.video_find_albums_album_videos_get import ( asyncio as video_find, ) from photosapi_client.api.default.album_find_albums_get import asyncio as album_find from photosapi_client.api.default.album_create_albums_post import ( asyncio as album_create, ) from photosapi_client.api.default.album_delete_album_id_delete import ( asyncio as album_delete, ) from photosapi_client.api.default.photo_find_albums_album_photos_get import ( asyncio as photo_find, ) from photosapi_client.api.default.photo_patch_photos_id_patch import ( asyncio as photo_patch, ) from photosapi_client.api.default.photo_delete_photos_id_delete import ( asyncio as photo_delete, ) from photosapi_client.api.default.photo_upload_albums_album_photos_post import ( asyncio_detailed as photo_upload, ) from photosapi_client.api.default.photo_get_photos_id_get import asyncio as photo_get from photosapi_client.models.body_photo_upload_albums_album_photos_post import ( BodyPhotoUploadAlbumsAlbumPhotosPost, ) from photosapi_client.types import File from photosapi_client.models.token import Token from photosapi_client.models.http_validation_error import HTTPValidationError from photosapi_client.models.body_login_for_access_token_token_post import ( BodyLoginForAccessTokenTokenPost, ) # import asyncio 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 import aiofiles # from aiohttp import FormData # from classes.exceptions import ( # AlbumCreationDuplicateError, # AlbumCreationError, # AlbumCreationNameError, # SubmissionUploadError, # UserCreationDuplicateError, # UserCreationError, # ) from modules.logger import logWrite from modules.utils import configGet, locale from modules.http_client import http_session async def authorize() -> str: makedirs(configGet("cache", "locations"), exist_ok=True) if path.exists(configGet("cache", "locations") + sep + "api_access") is True: async with aiofiles.open( configGet("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/", headers={"Authorization": f"Bearer {token}"}, ) ).status == 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 = await http_session.post( configGet("address", "posting", "api") + "/token", data=payload ) if not response.ok: logWrite( locale( "api_creds_invalid", "console", locale=configGet("locale_log").format( configGet("address", "posting", "api"), configGet("username", "posting", "api"), response.status, ), ) ) raise ValueError 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")) ) return (await response.json())["access_token"] unauthorized_client = Client( base_url=configGet("address", "posting", "api"), timeout=5.0, verify_ssl=True, raise_on_unexpected_status=True, ) 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=configGet("username", "posting", "api"), password=configGet("password", "posting", "api"), ), ) if not isinstance(login_token, Token): logWrite(f"Could not initialize connection due to invalid token: {login_token}") exit() client = AuthenticatedClient( base_url=configGet("address", "posting", "api"), timeout=5.0, verify_ssl=True, raise_on_unexpected_status=True, token=login_token.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={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, # ) # 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 if __name__ == "__main__": print(asyncio.run(authorize()))