TelegramPoster/modules/api_client.py

import asyncio
import logging
from base64 import b64decode, b64encode
from os import makedirs, path, sep
from pathlib import Path

import aiofiles
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 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_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.models.body_login_for_access_token_token_post import (
    BodyLoginForAccessTokenTokenPost,
)
from photosapi_client.models.body_photo_upload_albums_album_photos_post import (
    BodyPhotoUploadAlbumsAlbumPhotosPost,
)
from photosapi_client.models.http_validation_error import HTTPValidationError
from photosapi_client.models.token import Token
from photosapi_client.types import File

from modules.http_client import http_session

logger = logging.getLogger(__name__)


async def authorize() -> str:
    makedirs(await config_get("cache", "locations"), exist_ok=True)
    if path.exists(await config_get("cache", "locations") + sep + "api_access") is True:
        async with aiofiles.open(
            await config_get("cache", "locations") + sep + "api_access", "rb"
        ) as file:
            token = b64decode(await file.read()).decode("utf-8")
        if (
            await http_session.get(
                await config_get("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": await config_get("username", "posting", "api"),
        "password": await config_get("password", "posting", "api"),
    }
    response = await http_session.post(
        await config_get("address", "posting", "api") + "/token", data=payload
    )
    if not response.ok:
        logger.warning(
            i18n._(
                "api_creds_invalid",
                "console",
                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(
        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"))
        )
    return (await response.json())["access_token"]


unauthorized_client = Client(
    base_url=sync.config_get("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=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
    )
    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()))