78 lines
2.2 KiB
Python
78 lines
2.2 KiB
Python
import logging
|
|
import time
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from ftputil import FTPHost
|
|
from typing_extensions import override
|
|
|
|
from core.services.FtpService import FtpService
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class FtputilFtpService(FtpService):
|
|
|
|
domain: str
|
|
username: str
|
|
password: str
|
|
port: str
|
|
|
|
class Progress:
|
|
def __init__(self, size: int):
|
|
self.allBytes: int = size
|
|
self.transferedBytes: int = 0
|
|
self.startTime = time.time()
|
|
self.i = 0
|
|
|
|
def __call__(self, chunk: bytes):
|
|
self.i += 1
|
|
|
|
chunkBytes = len(chunk)
|
|
self.transferedBytes += chunkBytes
|
|
|
|
if self.i % 10 == 0:
|
|
duration = time.time() - self.startTime
|
|
progress = self.transferedBytes / self.allBytes * 100
|
|
numChunksLeft = (self.allBytes - self.transferedBytes) / chunkBytes
|
|
timeLeft = (duration * numChunksLeft) / 60
|
|
log.info(", ".join([
|
|
f"FTP progress: {round(progress, 1)}%",
|
|
f"Transferred: ({round(self.transferedBytes / 1e6, 1)} / {round(self.allBytes / 1e6, 1)}) MB",
|
|
f"Time left: {round(timeLeft, 2)} minutes"
|
|
]))
|
|
|
|
self.startTime = time.time()
|
|
|
|
def __post_init__(self):
|
|
self.ftp = FTPHost(self.domain, self.username, self.password, self.port)
|
|
|
|
@override
|
|
def download(self, path: Path):
|
|
log.info(f"Download: '{path.name}' to '{path}'")
|
|
# Download some files from the login directory.
|
|
self.ftp.download(source=path.name, target=path, callback=self.Progress(size=self.ftp.stat(path.name).st_size))
|
|
|
|
@override
|
|
def upload(self, path: Path):
|
|
log.info(f"Upload: '{path}' to '{path.name}'")
|
|
self.ftp.upload(source=path, target=path.name, callback=self.Progress(size=path.stat().st_size))
|
|
|
|
@override
|
|
def rename(self, oldPath: Path, newPath: Path):
|
|
log.info(f"Rename: '{oldPath.name}' to '{newPath.name}'")
|
|
self.ftp.rename(source=oldPath.name, target=newPath.name)
|
|
|
|
@override
|
|
def delete(self, path: Path):
|
|
log.info(f"Delete: '{path.name}'")
|
|
self.ftp.remove(path=path.name)
|
|
|
|
@override
|
|
def copy(self, path: Path, newPath: Path):
|
|
log.info(f"Copy: '{path.name}' to '{newPath.name}'")
|
|
with self.ftp.open(path.name, "rb") as source:
|
|
with self.ftp.open(newPath.name, "wb") as target:
|
|
self.ftp.copyfileobj(source=source, target=target)
|