import re from secrets import token_urlsafe from magic import Magic from datetime import datetime from os import makedirs, sep, path, remove from modules.hasher import get_phash, get_duplicates from modules.utils import configGet from modules.app import app, check_project_key, get_api_key from modules.database import col_photos, col_albums, col_tokens from bson.objectid import ObjectId from bson.errors import InvalidId from fastapi import HTTPException, Depends, UploadFile from fastapi.responses import UJSONResponse, Response from fastapi.openapi.models import APIKey from starlette.status import HTTP_204_NO_CONTENT, HTTP_400_BAD_REQUEST, HTTP_401_UNAUTHORIZED, HTTP_404_NOT_FOUND, HTTP_406_NOT_ACCEPTABLE, HTTP_409_CONFLICT @app.post("/albums/{album}/photos", response_class=UJSONResponse, include_in_schema=True) async def photo_upload(file: UploadFile, album: str, ignore_duplicates: bool = False, apikey: APIKey = Depends(get_api_key)): if (check_project_key("photos", apikey)): if col_albums.find_one( {"name": album} ) is None: return HTTPException(status_code=HTTP_404_NOT_FOUND, detail=f"Provided album '{album}' does not exist.") # if not file.content_type.startswith("image"): # return HTTPException(status_code=HTTP_406_NOT_ACCEPTABLE, detail="Provided file is not an image, not accepting.") makedirs(f'data{sep}users{sep}sample_user{sep}albums{sep}{album}', exist_ok=True) filename = file.filename if path.exists(f'data{sep}users{sep}sample_user{sep}albums{sep}{album}{sep}{file.filename}'): base_name = file.filename.split(".")[:-1] extension = file.filename.split(".")[-1] filename = ".".join(base_name)+f"_{int(datetime.now().timestamp())}."+extension with open(f'data{sep}users{sep}sample_user{sep}albums{sep}{album}{sep}{filename}', "wb") as f: f.write(await file.read()) file_hash = await get_phash(f'data{sep}users{sep}sample_user{sep}albums{sep}{album}{sep}{filename}') duplicates = await get_duplicates(file_hash, album) if len(duplicates) > 0 and ignore_duplicates is False: return UJSONResponse( { "detail": "Image duplicates found. Pass 'ignore_duplicates=true' to ignore.", "duplicates": duplicates }, status_code=HTTP_409_CONFLICT ) uploaded = col_photos.insert_one( {"album": album, "hash": file_hash, "filename": filename} ) return UJSONResponse( { "id": uploaded.inserted_id.__str__(), "album": album, "hash": file_hash, "filename": filename } ) else: raise HTTPException(status_code=HTTP_401_UNAUTHORIZED, detail=configGet("key_invalid", "messages")) @app.get("/photos/{id}", include_in_schema=True) async def photo_get(id: str, apikey: APIKey = Depends(get_api_key)): if (check_project_key("photos", apikey)): try: image = col_photos.find_one( {"_id": ObjectId(id)} ) if image is None: raise InvalidId(id) except InvalidId: return HTTPException(status_code=HTTP_404_NOT_FOUND, detail="Could not find an image with such id.") image_path = f'data{sep}users{sep}sample_user{sep}albums{sep}{image["album"]}{sep}{image["filename"]}' mime = Magic(mime=True).from_file(image_path) with open(image_path, "rb") as f: image_file = f.read() return Response(image_file, media_type=mime) else: raise HTTPException(status_code=HTTP_401_UNAUTHORIZED, detail=configGet("key_invalid", "messages")) @app.delete("/photos/{id}", include_in_schema=True) async def photo_delete(id: str, apikey: APIKey = Depends(get_api_key)): if (check_project_key("photos", apikey)): try: image = col_photos.find_one_and_delete( {"_id": ObjectId(id)} ) if image is None: raise InvalidId(id) except InvalidId: return HTTPException(status_code=HTTP_404_NOT_FOUND, detail="Could not find an image with such id.") remove(f'data{sep}users{sep}sample_user{sep}albums{sep}{image["album"]}{sep}{image["filename"]}') return Response(status_code=HTTP_204_NO_CONTENT) else: raise HTTPException(status_code=HTTP_401_UNAUTHORIZED, detail=configGet("key_invalid", "messages")) @app.get("/albums/{album}/photos", response_class=UJSONResponse, include_in_schema=True) async def photo_find(q: str, album: str, page: int = 1, page_size: int = 100, apikey: APIKey = Depends(get_api_key)): if (check_project_key("photos", apikey)): if col_albums.find_one( {"name": album} ) is None: return HTTPException(status_code=HTTP_404_NOT_FOUND, detail=f"Provided album '{album}' does not exist.") if page <= 0 or page_size <= 0: return HTTPException(status_code=HTTP_400_BAD_REQUEST, detail="Parameters 'page' and 'page_size' must be greater or equal to 1.") output = {"results": []} skip = (page-1)*page_size images = list(col_photos.find({"album": album, "filename": re.compile(q)}, limit=page_size, skip=skip)) for image in images: output["results"].append({"id": image["_id"].__str__(), "filename": image["filename"]}) if col_photos.count_documents( {"album": album, "filename": re.compile(q)} ) > page*page_size: token = str(token_urlsafe(32)) col_tokens.insert_one( {"token": token, "query": q, "album": album, "page": page+1, "page_size": page_size, "apikey": apikey} ) output["next_page"] = f"https://api.end-play.xyz/photoFindToken?token={token}" # type: ignore return UJSONResponse(output) else: raise HTTPException(status_code=HTTP_401_UNAUTHORIZED, detail=configGet("key_invalid", "messages")) @app.get("/photos/token/{token}", response_class=UJSONResponse, include_in_schema=True) async def photo_find_token(token: str): found_record = col_tokens.find_one( {"token": token} ) if found_record is None: return HTTPException(status_code=HTTP_401_UNAUTHORIZED, detail="Invalid search token.") return await photo_find(q=found_record["query"], album=found_record["album"], page=found_record["page"], page_size=found_record["page_size"], apikey=found_record["apikey"])