TelegramPoster/modules/api_client.py

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

import aiofiles
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 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_delete_videos_id_delete import (
    asyncio as video_delete,
)
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(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(
            await config_get("cache", "locations") + sep + "api_access", "rb"
        ) as file:
            token = b64decode(await file.read()).decode("utf-8")
        if (
            await 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 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()))