diff --git a/extensions/albums.py b/extensions/albums.py index 1eb130a..d7cecf3 100644 --- a/extensions/albums.py +++ b/extensions/albums.py @@ -1,5 +1,6 @@ import re -from os import makedirs, path, rename +from os import makedirs, rename +from pathlib import Path from shutil import rmtree from typing import Union @@ -49,9 +50,7 @@ async def album_create( if col_albums.find_one({"name": name}) is not None: raise AlbumAlreadyExistsError(name) - makedirs( - path.join("data", "users", current_user.user, "albums", name), exist_ok=True - ) + makedirs(Path(f"data/users/{current_user.user}/albums/{name}"), exist_ok=True) uploaded = col_albums.insert_one( {"user": current_user.user, "name": name, "title": title, "cover": None} @@ -123,8 +122,8 @@ async def album_patch( if 2 > len(name) > 20: raise AlbumIncorrectError("name", "must be >2 and <20 characters.") rename( - path.join("data", "users", current_user.user, "albums", album["name"]), - path.join("data", "users", current_user.user, "albums", name), + Path(f"data/users/{current_user.user}/albums/{album['name']}"), + Path(f"data/users/{current_user.user}/albums/{name}"), ) col_photos.update_many( {"user": current_user.user, "album": album["name"]}, @@ -186,8 +185,8 @@ async def album_put( cover = image["_id"].__str__() if image is not None else None # type: ignore rename( - path.join("data", "users", current_user.user, "albums", album["name"]), - path.join("data", "users", current_user.user, "albums", name), + Path(f"data/users/{current_user.user}/albums/{album['name']}"), + Path(f"data/users/{current_user.user}/albums/{name}"), ) col_photos.update_many( @@ -222,6 +221,6 @@ async def album_delete( col_photos.delete_many({"album": album["name"]}) - rmtree(path.join("data", "users", current_user.user, "albums", album["name"])) + rmtree(Path(f"data/users/{current_user.user}/albums/{album['name']}")) return Response(status_code=HTTP_204_NO_CONTENT) diff --git a/extensions/pages.py b/extensions/pages.py index d32dad6..b51f692 100644 --- a/extensions/pages.py +++ b/extensions/pages.py @@ -1,4 +1,4 @@ -from os import path +from pathlib import Path import aiofiles from fastapi.responses import HTMLResponse, Response @@ -8,27 +8,21 @@ from modules.app import app @app.get("/pages/matter.css", include_in_schema=False) async def page_matter(): - async with aiofiles.open( - path.join("pages", "matter.css"), "r", encoding="utf-8" - ) as f: + async with aiofiles.open(Path("pages/matter.css"), "r", encoding="utf-8") as f: output = await f.read() return Response(content=output) @app.get("/pages/{page}/{file}", include_in_schema=False) async def page_assets(page: str, file: str): - async with aiofiles.open( - path.join("pages", page, file), "r", encoding="utf-8" - ) as f: + async with aiofiles.open(Path(f"pages/{page}/{file}"), "r", encoding="utf-8") as f: output = await f.read() return Response(content=output) @app.get("/", include_in_schema=False) async def page_home(): - async with aiofiles.open( - path.join("pages", "home", "index.html"), "r", encoding="utf-8" - ) as f: + async with aiofiles.open(Path("pages/home/index.html"), "r", encoding="utf-8") as f: output = await f.read() return HTMLResponse(content=output) @@ -36,7 +30,7 @@ async def page_home(): @app.get("/register", include_in_schema=False) async def page_register(): async with aiofiles.open( - path.join("pages", "register", "index.html"), "r", encoding="utf-8" + Path("pages/register/index.html"), "r", encoding="utf-8" ) as f: output = await f.read() return HTMLResponse(content=output) diff --git a/extensions/photos.py b/extensions/photos.py index 902752a..1a0b338 100644 --- a/extensions/photos.py +++ b/extensions/photos.py @@ -1,6 +1,7 @@ import re from datetime import datetime, timedelta, timezone from os import makedirs, path, remove, system +from pathlib import Path from secrets import token_urlsafe from shutil import move from threading import Thread @@ -109,15 +110,11 @@ async def photo_upload( if col_albums.find_one({"user": current_user.user, "name": album}) is None: raise AlbumNameNotFoundError(album) - makedirs( - path.join("data", "users", current_user.user, "albums", album), exist_ok=True - ) + makedirs(Path(f"data/users/{current_user.user}/albums/{album}"), exist_ok=True) filename = file.filename - if path.exists( - path.join("data", "users", current_user.user, "albums", album, file.filename) - ): + if Path(f"data/users/{current_user.user}/albums/{album}/{file.filename}").exists(): base_name = file.filename.split(".")[:-1] extension = file.filename.split(".")[-1] filename = ( @@ -125,12 +122,12 @@ async def photo_upload( ) async with aiofiles.open( - path.join("data", "users", current_user.user, "albums", album, filename), "wb" + Path(f"data/users/{current_user.user}/albums/{album}/{filename}"), "wb" ) as f: await f.write(await file.read()) file_hash = await get_phash( - path.join("data", "users", current_user.user, "albums", album, filename) + Path(f"data/users/{current_user.user}/albums/{album}/{filename}") ) duplicates = await get_duplicates(file_hash, album) @@ -168,7 +165,7 @@ async def photo_upload( try: coords = extract_location( - path.join("data", "users", current_user.user, "albums", album, filename) + Path(f"data/users/{current_user.user}/albums/{album}/{filename}") ) except (UnpackError, ValueError): coords = {"lng": 0.0, "lat": 0.0, "alt": 0.0} @@ -193,9 +190,7 @@ async def photo_upload( compress_image, trigger="date", run_date=datetime.now() + timedelta(seconds=1), - args=[ - path.join("data", "users", current_user.user, "albums", album, filename) - ], + args=[Path(f"data/users/{current_user.user}/albums/{album}/{filename}")], ) return UJSONResponse( @@ -254,8 +249,8 @@ if configGet("media_token_access") is True: except InvalidId: raise PhotoNotFoundError(id) - image_path = path.join( - "data", "users", user.user, "albums", image["album"], image["filename"] + image_path = Path( + f"data/users/{user.user}/albums/{image['album']}/{image['filename']}" ) mime = Magic(mime=True).from_file(image_path) @@ -299,8 +294,8 @@ async def photo_get( except InvalidId: raise PhotoNotFoundError(id) - image_path = path.join( - "data", "users", current_user.user, "albums", image["album"], image["filename"] + image_path = Path( + f"data/users/{current_user.user}/albums/{image['album']}/{image['filename']}" ) mime = Magic(mime=True).from_file(image_path) @@ -335,11 +330,9 @@ async def photo_move( if col_albums.find_one({"user": current_user.user, "name": album}) is None: raise AlbumNameNotFoundError(album) - if path.exists( - path.join( - "data", "users", current_user.user, "albums", album, image["filename"] - ) - ): + if Path( + f"data/users/{current_user.user}/albums/{album}/{image['filename']}" + ).exists(): base_name = image["filename"].split(".")[:-1] extension = image["filename"].split(".")[-1] filename = ( @@ -360,15 +353,10 @@ async def photo_move( ) move( - path.join( - "data", - "users", - current_user.user, - "albums", - image["album"], - image["filename"], + Path( + f"data/users/{current_user.user}/albums/{image['album']}/{image['filename']}" ), - path.join("data", "users", current_user.user, "albums", album, filename), + Path(f"data/users/{current_user.user}/albums/{album}/{filename}"), ) return UJSONResponse( @@ -441,13 +429,8 @@ async def photo_delete( col_albums.update_one({"name": image["album"]}, {"$set": {"cover": None}}) remove( - path.join( - "data", - "users", - current_user.user, - "albums", - image["album"], - image["filename"], + Path( + f"data/users/{current_user.user}/albums/{image['album']}/{image['filename']}" ) ) diff --git a/extensions/videos.py b/extensions/videos.py index 2152d0a..415d0d4 100644 --- a/extensions/videos.py +++ b/extensions/videos.py @@ -5,6 +5,8 @@ from secrets import token_urlsafe from shutil import move from typing import Union +from pathlib import Path + import aiofiles from bson.errors import InvalidId from bson.objectid import ObjectId @@ -45,15 +47,11 @@ async def video_upload( if col_albums.find_one({"user": current_user.user, "name": album}) is None: raise AlbumNameNotFoundError(album) - makedirs( - path.join("data", "users", current_user.user, "albums", album), exist_ok=True - ) + makedirs(Path(f"data/users/{current_user.user}/albums/{album}"), exist_ok=True) filename = file.filename - if path.exists( - path.join("data", "users", current_user.user, "albums", album, file.filename) - ): + if Path(f"data/users/{current_user.user}/albums/{album}/{file.filename}").exists(): base_name = file.filename.split(".")[:-1] extension = file.filename.split(".")[-1] filename = ( @@ -61,7 +59,7 @@ async def video_upload( ) async with aiofiles.open( - path.join("data", "users", current_user.user, "albums", album, filename), "wb" + Path(f"data/users/{current_user.user}/albums/{album}/{filename}"), "wb" ) as f: await f.write(await file.read()) @@ -125,8 +123,8 @@ async def video_get( except InvalidId: raise VideoNotFoundError(id) - video_path = path.join( - "data", "users", current_user.user, "albums", video["album"], video["filename"] + video_path = Path( + f"data/users/{current_user.user}/albums/{video['album']}/{video['filename']}" ) mime = Magic(mime=True).from_file(video_path) @@ -161,11 +159,9 @@ async def video_move( if col_albums.find_one({"user": current_user.user, "name": album}) is None: raise AlbumNameNotFoundError(album) - if path.exists( - path.join( - "data", "users", current_user.user, "albums", album, video["filename"] - ) - ): + if Path( + f"data"/users/{current_user.user}/albums/{album}/{video['filename']}" + ).exists(): base_name = video["filename"].split(".")[:-1] extension = video["filename"].split(".")[-1] filename = ( @@ -186,15 +182,8 @@ async def video_move( ) move( - path.join( - "data", - "users", - current_user.user, - "albums", - video["album"], - video["filename"], - ), - path.join("data", "users", current_user.user, "albums", album, filename), + Path(f"data/users/{current_user.user}/albums/{video['album']}/{video['filename']}"), + Path(f"data/users/{current_user.user}/albums/{album}/{filename}"), ) return UJSONResponse( @@ -264,14 +253,7 @@ async def video_delete( album = col_albums.find_one({"name": video["album"]}) remove( - path.join( - "data", - "users", - current_user.user, - "albums", - video["album"], - video["filename"], - ) + Path(f"data/users/{current_user.user}/albums/{video['album']}/{video['filename']}") ) return Response(status_code=HTTP_204_NO_CONTENT) diff --git a/modules/extensions_loader.py b/modules/extensions_loader.py index e978fe2..30043e0 100644 --- a/modules/extensions_loader.py +++ b/modules/extensions_loader.py @@ -1,5 +1,6 @@ from importlib.util import module_from_spec, spec_from_file_location from os import getcwd, path, walk +from pathlib import Path # ================================================================================= @@ -12,7 +13,7 @@ def get_py_files(src): for root, dirs, files in walk(src): for file in files: if file.endswith(".py"): - py_files.append(path.join(cwd, root, file)) + py_files.append(Path(f"{cwd}/{root}/{file}")) return py_files diff --git a/photos_api.py b/photos_api.py index 714924b..b0142db 100644 --- a/photos_api.py +++ b/photos_api.py @@ -1,13 +1,13 @@ -from os import makedirs, path +from os import makedirs +from pathlib import Path from fastapi.responses import FileResponse from modules.app import app from modules.extensions_loader import dynamic_import_from_src from modules.scheduler import scheduler -from modules.utils import * -makedirs(path.join("data", "users"), exist_ok=True) +makedirs(Path("data/users"), exist_ok=True) @app.get("/favicon.ico", response_class=FileResponse, include_in_schema=False)