2 Commits

Author SHA1 Message Date
Profitroll
c272342b4b Fixed description 2023-02-18 00:47:21 +01:00
Profitroll
f6c2002811 Adjusted photo tokens 2023-02-18 00:47:00 +01:00

View File

@@ -4,6 +4,7 @@ from secrets import token_urlsafe
from shutil import move from shutil import move
from threading import Thread from threading import Thread
from typing import Union from typing import Union
from uuid import uuid4
from magic import Magic from magic import Magic
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from os import makedirs, path, remove, system from os import makedirs, path, remove, system
@@ -96,11 +97,14 @@ async def photo_upload(file: UploadFile, album: str, ignore_duplicates: bool = F
duplicates_ids = [] duplicates_ids = []
for entry in duplicates: for entry in duplicates:
duplicates_ids.append(entry["id"]) duplicates_ids.append(entry["id"])
access_token = create_access_token(data={"sub": current_user.user, "scopes": ["me", "photos.read"], "allowed": duplicates_ids}, expires_delta=timedelta(hours=1))
access_token_short = uuid4().hex[:12].lower()
col_tokens.insert_one({"short": access_token_short, "access_token": access_token, "photos": duplicates_ids})
return UJSONResponse( return UJSONResponse(
{ {
"detail": "Image duplicates found. Pass 'ignore_duplicates=true' to ignore.", "detail": "Image duplicates found. Pass 'ignore_duplicates=true' to ignore.",
"duplicates": duplicates, "duplicates": duplicates,
"access_token": create_access_token(data={"sub": current_user.user, "scopes": ["me", "photos.read"], "allowed": duplicates_ids}, expires_delta=timedelta(hours=1)) "access_token": access_token_short
}, },
status_code=HTTP_409_CONFLICT status_code=HTTP_409_CONFLICT
) )
@@ -149,8 +153,16 @@ photo_get_token_responses = {
401: AccessTokenInvalidError().openapi, 401: AccessTokenInvalidError().openapi,
404: PhotoNotFoundError("id").openapi 404: PhotoNotFoundError("id").openapi
} }
@app.get("/photos/{id}/token/{token}", description="Get a photo by id", responses=photo_get_token_responses) @app.get("/token/photo/{token}", description="Get a photo by its duplicate token", responses=photo_get_token_responses)
async def photo_get_token(id: str, token: str): async def photo_get_token(token: str, id: int):
db_entry = col_tokens.find_one({"short": token})
if db_entry is None:
raise AccessTokenInvalidError()
token = db_entry["access_token"]
id = db_entry["photos"][id]
try: try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])