97 lines
3.5 KiB
Python
97 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
from typing import BinaryIO
|
|
|
|
from zyrarbot.storage.upload_file_info import UploadFileInfo
|
|
|
|
|
|
class UploadsRepository:
|
|
"""Katalog z plikami wgranymi na serwer; nazwy zawsze sprowadzane do basename."""
|
|
|
|
def __init__(self, directory: str):
|
|
self._directory = directory
|
|
self._legacy_guild_id: int | None = None
|
|
|
|
@property
|
|
def directory(self) -> str:
|
|
return self._directory
|
|
|
|
def _guild_directory(self, guild_id: int | None) -> str:
|
|
if guild_id is None:
|
|
return self._directory
|
|
self._migrate_legacy(guild_id)
|
|
return os.path.join(self._directory, str(guild_id))
|
|
|
|
def _migrate_legacy(self, guild_id: int) -> None:
|
|
"""Przenosi stare, niesegregowane pliki do pierwszego użytego serwera."""
|
|
if self._legacy_guild_id is not None:
|
|
return
|
|
|
|
os.makedirs(self._directory, exist_ok=True)
|
|
legacy_files = [
|
|
name
|
|
for name in os.listdir(self._directory)
|
|
if os.path.isfile(os.path.join(self._directory, name)) and not name.startswith(".")
|
|
]
|
|
target = os.path.join(self._directory, str(guild_id))
|
|
if legacy_files:
|
|
os.makedirs(target, exist_ok=True)
|
|
for name in legacy_files:
|
|
shutil.move(
|
|
os.path.join(self._directory, name),
|
|
os.path.join(target, name),
|
|
)
|
|
self._legacy_guild_id = guild_id
|
|
|
|
def resolve(self, filename: str, guild_id: int | None = None) -> str:
|
|
return os.path.join(self._guild_directory(guild_id), os.path.basename(filename))
|
|
|
|
def exists(self, filename: str, guild_id: int | None = None) -> bool:
|
|
return os.path.isfile(self.resolve(filename, guild_id))
|
|
|
|
def list(self, guild_id: int | None = None) -> list[UploadFileInfo]:
|
|
directory = self._guild_directory(guild_id)
|
|
os.makedirs(directory, exist_ok=True)
|
|
files = []
|
|
for name in os.listdir(directory):
|
|
path = os.path.join(directory, name)
|
|
if not os.path.isfile(path) or name.startswith("."):
|
|
continue
|
|
files.append(
|
|
UploadFileInfo(
|
|
filename=name,
|
|
size_mb=round(os.path.getsize(path) / (1024 * 1024), 2),
|
|
mtime=os.path.getmtime(path),
|
|
)
|
|
)
|
|
files.sort(key=lambda item: item.mtime, reverse=True)
|
|
return files
|
|
|
|
def search(self, query: str, limit: int = 25, guild_id: int | None = None) -> list[str]:
|
|
needle = query.lower()
|
|
return [item.filename for item in self.list(guild_id) if needle in item.filename.lower()][
|
|
:limit
|
|
]
|
|
|
|
def save(self, filename: str, source: BinaryIO, guild_id: int | None = None) -> str:
|
|
directory = self._guild_directory(guild_id)
|
|
os.makedirs(directory, exist_ok=True)
|
|
name = os.path.basename(filename)
|
|
with open(self.resolve(name, guild_id), "wb") as target:
|
|
shutil.copyfileobj(source, target)
|
|
return name
|
|
|
|
def delete(self, filename: str, guild_id: int | None = None) -> bool:
|
|
path = self.resolve(filename, guild_id)
|
|
if not os.path.isfile(path):
|
|
return False
|
|
os.remove(path)
|
|
return True
|
|
|
|
def touch(self, filename: str, guild_id: int | None = None) -> None:
|
|
path = self.resolve(filename, guild_id)
|
|
if os.path.isfile(path):
|
|
os.utime(path, None)
|