Merge branch 'main' into linting_etc

# Conflicts:
#	src/auto_archiver/core/consts.py
#	src/auto_archiver/core/orchestrator.py
#	src/auto_archiver/core/storage.py
#	src/auto_archiver/modules/local_storage/local_storage.py
#	src/auto_archiver/modules/s3_storage/s3_storage.py
#	tests/storages/test_S3_storage.py
#	tests/storages/test_local_storage.py
#	tests/storages/test_storage_base.py
This commit is contained in:
erinhmclark
2025-03-11 10:39:47 +00:00
19 changed files with 614 additions and 320 deletions

View File

@@ -1,3 +1,4 @@
import shutil
from typing import IO
import os
@@ -5,25 +6,42 @@ from loguru import logger
from auto_archiver.core import Media
from auto_archiver.core import Storage
from auto_archiver.core.consts import SetupError
class LocalStorage(Storage):
def setup(self) -> None:
if len(self.save_to) > 200:
raise SetupError(f"Your save_to path is too long, this will cause issues saving files on your computer. Please use a shorter path.")
def get_cdn_url(self, media: Media) -> str:
# TODO: is this viable with Storage.configs on path/filename?
dest = os.path.join(self.save_to, media.key)
dest = media.key
if self.save_absolute:
dest = os.path.abspath(dest)
return dest
def set_key(self, media, url, metadata):
# clarify we want to save the file to the save_to folder
old_folder = metadata.get('folder', '')
metadata.set_context('folder', os.path.join(self.save_to, metadata.get('folder', '')))
super().set_key(media, url, metadata)
# don't impact other storages that might want a different 'folder' set
metadata.set_context('folder', old_folder)
def upload(self, media: Media, **kwargs) -> bool:
# override parent so that we can use shutil.copy2 and keep metadata
dest = os.path.join(self.save_to, media.key)
dest = media.key
os.makedirs(os.path.dirname(dest), exist_ok=True)
logger.debug(f"[{self.__class__.__name__}] storing file {media.filename} with key {media.key} to {dest}")
logger.debug(f'[{self.__class__.__name__}] storing file {media.filename} with key {media.key} to {dest}')
res = shutil.copy2(media.filename, dest)
logger.info(res)
return True
# must be implemented even if unused
def uploadf(self, file: IO[bytes], key: str, **kwargs: dict) -> bool:
pass
pass