dev #19
@ -1,3 +1,5 @@
|
|||||||
|
"""This is only a temporary solution. Complete Photos API client is yet to be developed."""
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
from base64 import b64decode, b64encode
|
from base64 import b64decode, b64encode
|
||||||
from os import makedirs, path, sep
|
from os import makedirs, path, sep
|
||||||
@ -37,8 +39,7 @@ async def random_pic(token: Union[str, None] = None) -> Tuple[str, str]:
|
|||||||
### Returns:
|
### Returns:
|
||||||
* `Tuple[str, str]`: First value is an ID and the filename in the filesystem to be indexed.
|
* `Tuple[str, str]`: First value is an ID and the filename in the filesystem to be indexed.
|
||||||
"""
|
"""
|
||||||
if token is None:
|
token = await authorize() if token is None else token
|
||||||
token = await authorize()
|
|
||||||
logWrite(f'{configGet("address", "posting", "api")}/albums/{configGet("album", "posting", "api")}/photos?q=&page_size=100&caption=queue')
|
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}"})
|
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:
|
if resp.status_code != 200:
|
||||||
@ -50,8 +51,8 @@ async def random_pic(token: Union[str, None] = None) -> Tuple[str, str]:
|
|||||||
pic = choice(resp.json()["results"])
|
pic = choice(resp.json()["results"])
|
||||||
return pic["id"], pic["filename"]
|
return pic["id"], pic["filename"]
|
||||||
|
|
||||||
async def upload_pic(filepath: str) -> Tuple[bool, list]:
|
async def upload_pic(filepath: str, token: Union[str, None] = None) -> Tuple[bool, list]:
|
||||||
token = await authorize()
|
token = await authorize() if token is None else token
|
||||||
try:
|
try:
|
||||||
pic_name = path.basename(filepath)
|
pic_name = path.basename(filepath)
|
||||||
files = {'file': (pic_name, open(filepath, 'rb'), 'image/jpeg')}
|
files = {'file': (pic_name, open(filepath, 'rb'), 'image/jpeg')}
|
||||||
@ -67,8 +68,22 @@ async def upload_pic(filepath: str) -> Tuple[bool, list]:
|
|||||||
except:
|
except:
|
||||||
return False, []
|
return False, []
|
||||||
|
|
||||||
async def move_pic(id: str) -> bool:
|
async def find_pic(name: str, caption: Union[str, None] = None, token: Union[str, None] = None) -> Union[dict, None]:
|
||||||
token = await authorize()
|
token = await authorize() if token is None else token
|
||||||
|
try:
|
||||||
|
response = 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_code != 200:
|
||||||
|
return None
|
||||||
|
if len(response.json()["results"]) == 0:
|
||||||
|
return None
|
||||||
|
return response.json()["results"]
|
||||||
|
except Exception as exp:
|
||||||
|
logWrite(f"Could not find image with name '{name}' and caption '{caption}' due to: {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:
|
try:
|
||||||
patch(f'{configGet("address", "posting", "api")}/photos/{id}?caption=sent', headers={"Authorization": f"Bearer {token}"})
|
patch(f'{configGet("address", "posting", "api")}/photos/{id}?caption=sent', headers={"Authorization": f"Bearer {token}"})
|
||||||
return True
|
return True
|
||||||
|
Reference in New Issue
Block a user