Spaces:
Running
Running
File size: 2,103 Bytes
09801ca | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 | import os
import shutil
from typing import Optional
# Base storage directory
STORAGE_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "storage", "user-files")
os.makedirs(STORAGE_DIR, exist_ok=True)
class LocalStorage:
def __init__(self):
self.available = True
def _get_user_dir(self, user_id: str) -> str:
d = os.path.join(STORAGE_DIR, user_id)
os.makedirs(d, exist_ok=True)
return d
def upload_file(self, user_id: str, filename: str, file_data: bytes, content_type: str = "application/octet-stream") -> dict:
user_dir = self._get_user_dir(user_id)
filepath = os.path.join(user_dir, filename)
try:
with open(filepath, "wb") as f:
f.write(file_data)
return {"success": True, "path": f"{user_id}/{filename}", "response": {}}
except Exception as e:
return {"success": False, "error": str(e)}
def download_file(self, user_id: str, filename: str) -> bytes:
filepath = os.path.join(self._get_user_dir(user_id), filename)
with open(filepath, "rb") as f:
return f.read()
def delete_file(self, user_id: str, filename: str) -> dict:
filepath = os.path.join(self._get_user_dir(user_id), filename)
try:
if os.path.exists(filepath):
os.remove(filepath)
return {"success": True, "response": {}}
except Exception as e:
return {"success": False, "error": str(e)}
def list_files(self, user_id: str) -> list:
user_dir = self._get_user_dir(user_id)
try:
return [{"name": f} for f in os.listdir(user_dir) if os.path.isfile(os.path.join(user_dir, f))]
except:
return []
def get_public_url(self, user_id: str, filename: str) -> str:
return f"/api/v1/files/download/{user_id}/{filename}"
def get_signed_url(self, user_id: str, filename: str, expires_in: int = 3600) -> str:
return self.get_public_url(user_id, filename)
def get_storage() -> LocalStorage:
return LocalStorage()
|