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

77 lines
3.8 KiB
Python
Raw Normal View History

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-02-14 17:25:56 +02:00
from typing import Tuple, Union
from requests import get, patch, post
from modules.logger import logWrite
2023-01-17 15:11:23 +02:00
from modules.utils import configGet
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)
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")
2023-02-14 17:25:56 +02:00
if get(configGet("address", "posting", "api")+"/users/me/", headers={"Authorization": f"Bearer {token}"}).status_code == 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"),
"password": configGet("password", "posting", "api")
}
2023-02-14 17:25:56 +02:00
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
2023-01-17 15:11:23 +02:00
with open(configGet("cache", "locations")+sep+"api_access", "wb") as file:
2023-02-14 17:25:56 +02:00
file.write(b64encode(response.json()["access_token"].encode("utf-8")))
return response.json()["access_token"]
2023-01-17 15:11:23 +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-02-14 17:25:56 +02:00
if token is None:
token = await authorize()
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"])
2023-01-17 15:11:23 +02:00
return pic["id"], pic["filename"]
2023-02-14 17:25:56 +02:00
async def upload_pic(filepath: str) -> Tuple[bool, list]:
token = await authorize()
2023-01-17 15:11:23 +02:00
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&caption=queue', headers={"Authorization": f"Bearer {token}"}, files=files).json()
print(response, flush=True)
duplicates = []
if "duplicates" in response:
for duplicate in response["duplicates"]:
duplicates.append(f'{configGet("address", "posting", "api")}/photos/{duplicate["id"]}')
return True, duplicates
except:
return False, []
2023-02-14 17:25:56 +02:00
async def move_pic(id: str) -> bool:
token = await authorize()
2023-01-17 15:11:23 +02:00
try:
patch(f'{configGet("address", "posting", "api")}/photos/{id}?caption=sent', headers={"Authorization": f"Bearer {token}"})
return True
except:
return False
if __name__ == "__main__":
2023-02-14 17:25:56 +02:00
print(asyncio.run(authorize()))