dinlo
f624fee5e0
- SSH/SFTP управление без агентов - Пакеты (apt), Docker, файлы, systemd, ufw, cron - Плейбуки, групповые операции, планировщик - Бэкапы, отчёты, Portainer, Proxmox VE, Gitea Deploy - Терминал, Telegram-уведомления, тёмная/светлая тема
124 lines
3.8 KiB
Python
124 lines
3.8 KiB
Python
"""SFTP файловый менеджер."""
|
|
from __future__ import annotations
|
|
|
|
import stat
|
|
from dataclasses import dataclass
|
|
from pathlib import PurePosixPath
|
|
from typing import Callable, List, Optional
|
|
|
|
import asyncssh
|
|
|
|
from app.core.ssh import SSHClient
|
|
|
|
|
|
@dataclass
|
|
class FileEntry:
|
|
name: str
|
|
path: str
|
|
is_dir: bool
|
|
size: int
|
|
mtime: float
|
|
permissions: str
|
|
|
|
|
|
def _fmt_perms(mode: int) -> str:
|
|
return stat.filemode(mode) if mode else "?"
|
|
|
|
|
|
async def list_dir(c: SSHClient, path: str) -> List[FileEntry]:
|
|
async with c.conn.start_sftp_client() as sftp:
|
|
entries = []
|
|
try:
|
|
attrs = await sftp.readdir(path)
|
|
except asyncssh.SFTPFailure as e:
|
|
raise PermissionError(str(e))
|
|
for a in attrs:
|
|
if a.filename in (".", ".."):
|
|
continue
|
|
full = str(PurePosixPath(path) / a.filename)
|
|
is_dir = stat.S_ISDIR(a.attrs.permissions or 0)
|
|
entries.append(
|
|
FileEntry(
|
|
name=a.filename,
|
|
path=full,
|
|
is_dir=is_dir,
|
|
size=a.attrs.size or 0,
|
|
mtime=a.attrs.mtime or 0,
|
|
permissions=_fmt_perms(a.attrs.permissions or 0),
|
|
)
|
|
)
|
|
entries.sort(key=lambda e: (not e.is_dir, e.name.lower()))
|
|
return entries
|
|
|
|
|
|
async def download(c: SSHClient, remote: str, local: str,
|
|
progress: Optional[Callable[[int, int], None]] = None) -> None:
|
|
async with c.conn.start_sftp_client() as sftp:
|
|
size = (await sftp.stat(remote)).size or 0
|
|
done = 0
|
|
async with sftp.open(remote, "rb") as rf, open(local, "wb") as lf:
|
|
while True:
|
|
chunk = await rf.read(65536)
|
|
if not chunk:
|
|
break
|
|
lf.write(chunk)
|
|
done += len(chunk)
|
|
if progress:
|
|
progress(done, size)
|
|
|
|
|
|
async def upload(c: SSHClient, local: str, remote: str,
|
|
progress: Optional[Callable[[int, int], None]] = None) -> None:
|
|
import os
|
|
size = os.path.getsize(local)
|
|
done = 0
|
|
async with c.conn.start_sftp_client() as sftp:
|
|
async with sftp.open(remote, "wb") as rf, open(local, "rb") as lf:
|
|
while True:
|
|
chunk = lf.read(65536)
|
|
if not chunk:
|
|
break
|
|
await rf.write(chunk)
|
|
done += len(chunk)
|
|
if progress:
|
|
progress(done, size)
|
|
|
|
|
|
async def delete(c: SSHClient, path: str, is_dir: bool) -> None:
|
|
async with c.conn.start_sftp_client() as sftp:
|
|
if is_dir:
|
|
await c.run(f"rm -rf {path!r}")
|
|
else:
|
|
await sftp.remove(path)
|
|
|
|
|
|
async def mkdir(c: SSHClient, path: str) -> None:
|
|
async with c.conn.start_sftp_client() as sftp:
|
|
await sftp.makedirs(path, exist_ok=True)
|
|
|
|
|
|
async def rename(c: SSHClient, old: str, new: str) -> None:
|
|
async with c.conn.start_sftp_client() as sftp:
|
|
await sftp.rename(old, new)
|
|
|
|
|
|
async def read_text(c: SSHClient, path: str, limit: int = 500_000) -> str:
|
|
async with c.conn.start_sftp_client() as sftp:
|
|
async with sftp.open(path, "rb") as f:
|
|
data = await f.read(limit)
|
|
return data.decode("utf-8", errors="replace")
|
|
|
|
|
|
async def write_text(c: SSHClient, path: str, content: str, use_sudo: bool = False) -> None:
|
|
if use_sudo:
|
|
# запись в root-owned файл через sudo tee
|
|
import base64
|
|
b64 = base64.b64encode(content.encode()).decode()
|
|
r = await c.sudo(f"echo {b64} | base64 -d > {path!r}")
|
|
if not r.ok:
|
|
raise PermissionError(r.stderr)
|
|
else:
|
|
async with c.conn.start_sftp_client() as sftp:
|
|
async with sftp.open(path, "wb") as f:
|
|
await f.write(content.encode())
|