112 lines
3.4 KiB
Python
112 lines
3.4 KiB
Python
import os
|
|
from .base import AbstractFS
|
|
|
|
|
|
class LocalFS(AbstractFS):
|
|
|
|
PLUGIN_NAME = "local"
|
|
PLUGIN_LABEL = "Local"
|
|
|
|
def __init__(self):
|
|
self._root = "/"
|
|
self._connected = False
|
|
|
|
def connect(self, config: dict):
|
|
self._root = config.get("root_path", "/")
|
|
self._connected = True
|
|
|
|
def disconnect(self):
|
|
self._connected = False
|
|
|
|
def is_connected(self) -> bool:
|
|
return self._connected
|
|
|
|
# ─── Navigation ──────────────────────────────────────────────
|
|
|
|
def list(self, path: str) -> list:
|
|
entries = sorted(os.scandir(path), key=lambda e: (not e.is_dir(), e.name.lower()))
|
|
items = []
|
|
for entry in entries:
|
|
try:
|
|
stat = entry.stat()
|
|
items.append({
|
|
'name': entry.name,
|
|
'path': entry.path,
|
|
'is_dir': entry.is_dir(),
|
|
'size': stat.st_size if not entry.is_dir() else 0,
|
|
'mtime': stat.st_mtime,
|
|
})
|
|
except PermissionError:
|
|
continue
|
|
return items
|
|
|
|
def isdir(self, path: str) -> bool:
|
|
return os.path.isdir(path)
|
|
|
|
def exists(self, path: str) -> bool:
|
|
return os.path.exists(path)
|
|
|
|
def getsize(self, path: str) -> int:
|
|
return os.path.getsize(path)
|
|
|
|
def join(self, *parts) -> str:
|
|
return os.path.join(*parts)
|
|
|
|
def basename(self, path: str) -> str:
|
|
return os.path.basename(path)
|
|
|
|
def dirname(self, path: str) -> str:
|
|
return os.path.dirname(path)
|
|
|
|
def relpath(self, path: str, base: str) -> str:
|
|
return os.path.relpath(path, base)
|
|
|
|
# ─── Opérations ──────────────────────────────────────────────
|
|
|
|
def mkdir(self, path: str):
|
|
os.makedirs(path, exist_ok=True)
|
|
|
|
def rename(self, old_path: str, new_path: str):
|
|
os.rename(old_path, new_path)
|
|
|
|
def remove(self, path: str):
|
|
if os.path.isdir(path):
|
|
import shutil
|
|
shutil.rmtree(path)
|
|
else:
|
|
os.remove(path)
|
|
|
|
def walk(self, path: str):
|
|
yield from os.walk(path)
|
|
|
|
# ─── Transfert ───────────────────────────────────────────────
|
|
|
|
def read_chunks(self, path: str, chunk_size: int = 4 * 1024 * 1024):
|
|
with open(path, 'rb') as f:
|
|
while True:
|
|
buf = f.read(chunk_size)
|
|
if not buf:
|
|
break
|
|
yield buf
|
|
|
|
def write_chunks(self, path: str, chunks):
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
with open(path, 'wb') as f:
|
|
for chunk in chunks:
|
|
f.write(chunk)
|
|
|
|
# ─── Config ──────────────────────────────────────────────────
|
|
|
|
@classmethod
|
|
def get_config_fields(cls) -> list:
|
|
return [
|
|
{
|
|
'name': 'root_path',
|
|
'label': 'Chemin racine',
|
|
'type': 'text',
|
|
'required': True,
|
|
'default': '/',
|
|
'placeholder': '/mnt/nas'
|
|
}
|
|
]
|