Merge pull request 'Random media requests' (#18) from dev into master

Reviewed-on: #18

classes/exceptions.py

@@ -164,6 +164,26 @@ class VideoSearchQueryEmptyError(HTTPException):
         )
 
 
+class SearchLimitInvalidError(HTTPException):
+    """Raises HTTP 400 if search results limit not in valid range."""
+
+    def __init__(self):
+        self.openapi = {
+            "description": "Invalid Limit",
+            "content": {
+                "application/json": {
+                    "example": {
+                        "detail": "Parameter 'limit' must be greater or equal to 1."
+                    }
+                }
+            },
+        }
+        super().__init__(
+            status_code=400,
+            detail=self.openapi["content"]["application/json"]["example"]["detail"],
+        )
+
+
 class SearchPageInvalidError(HTTPException):
     """Raises HTTP 400 if page or page size are not in valid range."""
 

classes/models.py

@@ -72,3 +72,11 @@ class SearchResultsPhoto(BaseModel):
 class SearchResultsVideo(BaseModel):
     results: List[VideoSearch]
     next_page: Union[str, None]
+
+
+class RandomSearchResultsPhoto(BaseModel):
+    results: List[PhotoSearch]
+
+
+class RandomSearchResultsVideo(BaseModel):
+    results: List[VideoSearch]

extensions/albums.py

@@ -108,12 +108,12 @@ async def album_patch(
     except InvalidId:
         raise AlbumNotFoundError(id)
 
-    if title is not None:
-        if 2 > len(title) > 40:
-            raise AlbumIncorrectError("title", "must be >2 and <40 characters.")
-    else:
+    if title is None:
         title = album["title"]
 
+    elif 2 > len(title) > 40:
+        raise AlbumIncorrectError("title", "must be >2 and <40 characters.")
+
     if name is not None:
         if re.search(re.compile("^[a-z,0-9,_]*$"), name) is False:
             raise AlbumIncorrectError(

extensions/photos.py

@@ -3,6 +3,7 @@ import re
 from datetime import datetime, timedelta, timezone
 from os import makedirs, path, remove, system
 from pathlib import Path
+from random import randint
 from secrets import token_urlsafe
 from shutil import move
 from threading import Thread
@@ -26,10 +27,16 @@ from classes.exceptions import (
     AlbumNameNotFoundError,
     PhotoNotFoundError,
     PhotoSearchQueryEmptyError,
+    SearchLimitInvalidError,
     SearchPageInvalidError,
     SearchTokenInvalidError,
 )
-from classes.models import Photo, PhotoPublic, SearchResultsPhoto
+from classes.models import (
+    Photo,
+    PhotoPublic,
+    RandomSearchResultsPhoto,
+    SearchResultsPhoto,
+)
 from modules.app import app
 from modules.database import col_albums, col_photos, col_tokens
 from modules.exif_reader import extract_location
@@ -139,11 +146,9 @@ async def photo_upload(
     )
     duplicates = await get_duplicates(file_hash, album)
 
-    if len(duplicates) > 0 and ignore_duplicates is False:
+    if len(duplicates) > 0 and not ignore_duplicates:
         if configGet("media_token_access") is True:
-            duplicates_ids = []
-            for entry in duplicates:
-                duplicates_ids.append(entry["id"])
+            duplicates_ids = [entry["id"] for entry in duplicates]
             access_token = create_access_token(
                 data={
                     "sub": current_user.user,
@@ -193,7 +198,7 @@ async def photo_upload(
         }
     )
 
-    if compress is True:
+    if compress:
         scheduler.add_job(
             compress_image,
             trigger="date",
@@ -445,6 +450,71 @@ async def photo_delete(
     return Response(status_code=HTTP_204_NO_CONTENT)
 
 
+photo_random_responses = {
+    400: SearchLimitInvalidError().openapi,
+    404: AlbumNameNotFoundError("name").openapi,
+}
+
+
+@app.get(
+    "/albums/{album}/photos/random",
+    description="Get one random photo, optionally by caption",
+    response_class=UJSONResponse,
+    response_model=RandomSearchResultsPhoto,
+    responses=photo_random_responses,
+)
+async def photo_random(
+    album: str,
+    caption: Union[str, None] = None,
+    limit: int = 100,
+    current_user: User = Security(get_current_active_user, scopes=["photos.list"]),
+):
+    if col_albums.find_one({"user": current_user.user, "name": album}) is None:
+        raise AlbumNameNotFoundError(album)
+
+    if limit <= 0:
+        raise SearchLimitInvalidError()
+
+    output = {"results": []}
+
+    db_query = (
+        {
+            "user": current_user.user,
+            "album": album,
+            "caption": re.compile(caption),
+        }
+        if caption is not None
+        else {
+            "user": current_user.user,
+            "album": album,
+        }
+    )
+
+    documents_count = col_photos.count_documents(db_query)
+    skip = randint(0, documents_count - 1) if documents_count > 1 else 0
+
+    images = list(
+        col_photos.aggregate(
+            [
+                {"$match": db_query},
+                {"$skip": skip},
+                {"$limit": limit},
+            ]
+        )
+    )
+
+    for image in images:
+        output["results"].append(
+            {
+                "id": image["_id"].__str__(),
+                "filename": image["filename"],
+                "caption": image["caption"],
+            }
+        )
+
+    return UJSONResponse(output)
+
+
 photo_find_responses = {
     400: SearchPageInvalidError().openapi,
     401: SearchTokenInvalidError().openapi,
@@ -519,7 +589,7 @@ async def photo_find(
         }
     elif q is None and caption is None:
         raise PhotoSearchQueryEmptyError()
-    elif q is None and caption is not None:
+    elif q is None:
         db_query = {
             "user": current_user.user,
             "album": album,
@@ -530,7 +600,7 @@ async def photo_find(
             "album": album,
             "caption": re.compile(caption),
         }
-    elif q is not None and caption is None:
+    elif caption is None:
         db_query = {
             "user": current_user.user,
             "album": album,

extensions/videos.py

@@ -2,6 +2,7 @@ import re
 from datetime import datetime, timezone
 from os import makedirs, remove
 from pathlib import Path
+from random import randint
 from secrets import token_urlsafe
 from shutil import move
 from typing import Union
@@ -17,12 +18,18 @@ from starlette.status import HTTP_204_NO_CONTENT
 
 from classes.exceptions import (
     AlbumNameNotFoundError,
+    SearchLimitInvalidError,
     SearchPageInvalidError,
     SearchTokenInvalidError,
     VideoNotFoundError,
     VideoSearchQueryEmptyError,
 )
-from classes.models import SearchResultsVideo, Video, VideoPublic
+from classes.models import (
+    RandomSearchResultsVideo,
+    SearchResultsVideo,
+    Video,
+    VideoPublic,
+)
 from modules.app import app
 from modules.database import col_albums, col_tokens, col_videos
 from modules.security import User, get_current_active_user
@@ -262,6 +269,71 @@ async def video_delete(
     return Response(status_code=HTTP_204_NO_CONTENT)
 
 
+video_random_responses = {
+    400: SearchLimitInvalidError().openapi,
+    404: AlbumNameNotFoundError("name").openapi,
+}
+
+
+@app.get(
+    "/albums/{album}/videos/random",
+    description="Get one random video, optionally by caption",
+    response_class=UJSONResponse,
+    response_model=RandomSearchResultsVideo,
+    responses=video_random_responses,
+)
+async def video_random(
+    album: str,
+    caption: Union[str, None] = None,
+    limit: int = 100,
+    current_user: User = Security(get_current_active_user, scopes=["videos.list"]),
+):
+    if col_albums.find_one({"user": current_user.user, "name": album}) is None:
+        raise AlbumNameNotFoundError(album)
+
+    if limit <= 0:
+        raise SearchLimitInvalidError()
+
+    output = {"results": []}
+
+    db_query = (
+        {
+            "user": current_user.user,
+            "album": album,
+            "caption": re.compile(caption),
+        }
+        if caption is not None
+        else {
+            "user": current_user.user,
+            "album": album,
+        }
+    )
+
+    documents_count = col_videos.count_documents(db_query)
+    skip = randint(0, documents_count - 1) if documents_count > 1 else 0
+
+    videos = list(
+        col_videos.aggregate(
+            [
+                {"$match": db_query},
+                {"$skip": skip},
+                {"$limit": limit},
+            ]
+        )
+    )
+
+    for video in videos:
+        output["results"].append(
+            {
+                "id": video["_id"].__str__(),
+                "filename": video["filename"],
+                "caption": video["caption"],
+            }
+        )
+
+    return UJSONResponse(output)
+
+
 video_find_responses = {
     400: SearchPageInvalidError().openapi,
     401: SearchTokenInvalidError().openapi,
@@ -313,7 +385,7 @@ async def video_find(
     if q is None and caption is None:
         raise VideoSearchQueryEmptyError()
 
-    if q is None and caption is not None:
+    if q is None:
         db_query = {
             "user": current_user.user,
             "album": album,
@@ -324,7 +396,7 @@ async def video_find(
             "album": album,
             "caption": re.compile(caption),
         }
-    elif q is not None and caption is None:
+    elif caption is None:
         db_query = list(
             col_videos.find(
                 {"user": current_user.user, "album": album, "filename": re.compile(q)},

modules/app.py

@@ -1,14 +1,14 @@
 from fastapi import FastAPI
 from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html
 
-app = FastAPI(title="END PLAY Photos", docs_url=None, redoc_url=None, version="0.4")
+app = FastAPI(title="END PLAY Photos", docs_url=None, redoc_url=None, version="0.5")
 
 
 @app.get("/docs", include_in_schema=False)
 async def custom_swagger_ui_html():
     return get_swagger_ui_html(
-        openapi_url=app.openapi_url,  # type: ignore
-        title=app.title + " - Documentation",
+        openapi_url=app.openapi_url,
+        title=f"{app.title} - Documentation",
         swagger_favicon_url="/favicon.ico",
     )
 
@@ -16,7 +16,7 @@ async def custom_swagger_ui_html():
 @app.get("/redoc", include_in_schema=False)
 async def custom_redoc_html():
     return get_redoc_html(
-        openapi_url=app.openapi_url,  # type: ignore
-        title=app.title + " - Documentation",
+        openapi_url=app.openapi_url,
+        title=f"{app.title} - Documentation",
         redoc_favicon_url="/favicon.ico",
     )

modules/database.py

@@ -24,7 +24,7 @@ db = db_client.get_database(name=db_config["name"])
 collections = db.list_collection_names()
 
 for collection in ["users", "albums", "photos", "videos", "tokens", "emails"]:
-    if not collection in collections:
+    if collection not in collections:
         db.create_collection(collection)
 
 col_users = db.get_collection("users")

modules/exif_reader.py

@@ -1,3 +1,5 @@
+import contextlib
+
 from exif import Image
 
 
@@ -12,8 +14,10 @@ def decimal_coords(coords: float, ref: str) -> float:
         * float: Decimal degrees
     """
     decimal_degrees = coords[0] + coords[1] / 60 + coords[2] / 3600
-    if ref == "S" or ref == "W":
+
+    if ref in {"S", "W"}:
         decimal_degrees = -decimal_degrees
+
     return round(decimal_degrees, 5)
 
 
@@ -35,11 +39,9 @@ def extract_location(filepath: str) -> dict:
     if img.has_exif is False:
         return output
 
-    try:
+    with contextlib.suppress(AttributeError):
         output["lng"] = decimal_coords(img.gps_longitude, img.gps_longitude_ref)
         output["lat"] = decimal_coords(img.gps_latitude, img.gps_latitude_ref)
         output["alt"] = img.gps_altitude
-    except AttributeError:
-        pass
 
     return output

modules/extensions_loader.py

@@ -11,9 +11,9 @@ def get_py_files(src):
     cwd = getcwd()  # Current Working directory
     py_files = []
     for root, dirs, files in walk(src):
-        for file in files:
-            if file.endswith(".py"):
-                py_files.append(Path(f"{cwd}/{root}/{file}"))
+        py_files.extend(
+            Path(f"{cwd}/{root}/{file}") for file in files if file.endswith(".py")
+        )
     return py_files
 
 

modules/hasher.py

@@ -1,3 +1,6 @@
+from pathlib import Path
+from typing import Union
+
 import cv2
 import numpy as np
 from numpy.typing import NDArray
@@ -17,18 +20,18 @@ def hash_hex_to_hash_array(hash_hex) -> NDArray:
     # convert hash string in hex to hash values of 0 or 1
     hash_str = int(hash_hex, 16)
     array_str = bin(hash_str)[2:]
-    return np.array([i for i in array_str], dtype=np.float32)
+    return np.array(list(array_str), dtype=np.float32)
 
 
 def get_duplicates_cache(album: str) -> dict:
-    output = {}
-    for photo in col_photos.find({"album": album}):
-        output[photo["filename"]] = [photo["_id"].__str__(), photo["hash"]]
-    return output
+    return {
+        photo["filename"]: [photo["_id"].__str__(), photo["hash"]]
+        for photo in col_photos.find({"album": album})
+    }
 
 
-async def get_phash(filepath: str) -> str:
-    img = cv2.imread(filepath)
+async def get_phash(filepath: Union[str, Path]) -> str:
+    img = cv2.imread(str(filepath))
     # resize image and convert to gray scale
     img = cv2.resize(img, (64, 64))
     img = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
@@ -49,14 +52,14 @@ async def get_phash(filepath: str) -> str:
     return hash_array_to_hash_hex(dct_block.flatten())
 
 
-async def get_duplicates(hash: str, album: str) -> list:
+async def get_duplicates(hash_string: str, album: str) -> list:
     duplicates = []
     cache = get_duplicates_cache(album)
-    for image_name in cache.keys():
+    for image_name, image_object in cache.items():
         try:
             distance = spatial.distance.hamming(
                 hash_hex_to_hash_array(cache[image_name][1]),
-                hash_hex_to_hash_array(hash),
+                hash_hex_to_hash_array(hash_string),
             )
         except ValueError:
             continue

modules/security.py

@@ -73,12 +73,10 @@ def get_user(user: str):
 
 
 def authenticate_user(user_name: str, password: str):
-    user = get_user(user_name)
-    if not user:
+    if user := get_user(user_name):
+        return user if verify_password(password, user.hash) else False
+    else:
         return False
-    if not verify_password(password, user.hash):
-        return False
-    return user
 
 
 def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
@@ -89,9 +87,8 @@ def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None
         expire = datetime.now(tz=timezone.utc) + timedelta(
             days=ACCESS_TOKEN_EXPIRE_DAYS
         )
-    to_encode.update({"exp": expire})
-    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
-    return encoded_jwt
+    to_encode["exp"] = expire
+    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
 
 
 async def get_current_user(

requirements.txt

@@ -7,5 +7,5 @@ passlib~=1.7.4
 pymongo==4.4.0
 python-jose[cryptography]~=3.3.0
 python-magic~=0.4.27
-scipy~=1.10.1
+scipy~=1.11.0
 ujson~=5.8.0