profitroll/TelegramPoster
Archived
1
0
Fork 0
Code Issues Pull Requests Releases 3 Wiki Activity

WIP: API usage as main

Browse Source
This commit is contained in:
Profitroll
2023-02-14 16:25:56 +01:00
parent 8e0fee4cb9
commit 7940cbe5a7
7 changed files with 302 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
modules/api_client.py
View File

@@ -1,16 +1,21 @@
import asyncio
from base64 import b64decode, b64encode
from os import makedirs, path, sep
from base64 import b64encode, b64decode
from random import choice
from typing import Tuple
from requests import get, post, put, patch
from typing import Tuple, Union
from requests import get, patch, post
from modules.logger import logWrite
from modules.utils import configGet
def authorize() -> str:
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 "user" in get(configGet("address", "posting", "api")+"/users/me/", headers={"Authorization": f"Bearer {token}"}).json():
if get(configGet("address", "posting", "api")+"/users/me/", headers={"Authorization": f"Bearer {token}"}).status_code == 200:
return token
payload = {
"grant_type": "password",
@@ -18,23 +23,35 @@ def authorize() -> str:
"username": configGet("username", "posting", "api"),
"password": configGet("password", "posting", "api")
}
response = post(configGet("address", "posting", "api")+"/token", data=payload).json()
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["access_token"].encode("utf-8")))
return response["access_token"]
file.write(b64encode(response.json()["access_token"].encode("utf-8")))
return response.json()["access_token"]
def random_pic() -> Tuple[str, str]:
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 = authorize()
pic = choice(get(f'{configGet("address", "posting", "api")}/albums/{configGet("album", "posting", "api")}/photos?q=&page_size=100&caption=queue', headers={"Authorization": f"Bearer {token}"}).json()["results"])
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"])
return pic["id"], pic["filename"]
def upload_pic(filepath: str) -> bool:
token = authorize()
async def upload_pic(filepath: str) -> Tuple[bool, list]:
token = await authorize()
try:
pic_name = path.basename(filepath)
files = {'file': (pic_name, open(filepath, 'rb'), 'image/jpeg')}
@@ -48,8 +65,8 @@ def upload_pic(filepath: str) -> bool:
except:
return False, []
def move_pic(id: str) -> bool:
token = authorize()
async def move_pic(id: str) -> bool:
token = await authorize()
try:
patch(f'{configGet("address", "posting", "api")}/photos/{id}?caption=sent', headers={"Authorization": f"Bearer {token}"})
return True
@@ -57,4 +74,4 @@ def move_pic(id: str) -> bool:
return False
if __name__ == "__main__":
print(authorize())
print(asyncio.run(authorize()))