use std::io::Cursor;
use std::path::{Path, PathBuf};
use std::time::SystemTime;
use async_ftp::FtpStream;
use async_trait::async_trait;
use tokio::io::AsyncReadExt;
use crate::error::{Error, Result};
use crate::traits::RemoteFileSystem;
use crate::types::{FileEntry, FileMeta, FtpConfig};
pub struct FtpClient {
config: FtpConfig,
stream: Option<FtpStream>,
}
impl FtpClient {
pub fn new(config: FtpConfig) -> Self {
Self {
config,
stream: None,
}
}
fn stream(&mut self) -> Result<&mut FtpStream> {
self.stream.as_mut().ok_or(Error::NotConnected)
}
fn addr(&self) -> String {
format!("{}:{}", self.config.host, self.config.port)
}
}
fn parse_list_line(base: &str, line: &str) -> Option<FileEntry> {
let line = line.trim();
if line.is_empty() {
return None;
}
if line.starts_with('d')
|| line.starts_with('-')
|| line.starts_with('l')
|| line.starts_with('b')
|| line.starts_with('c')
{
let parts: Vec<&str> = line.split_whitespace().collect();
if parts.len() < 9 {
return None;
}
let name = parts[8..].join(" ");
if name == "." || name == ".." {
return None;
}
let is_dir = parts[0].starts_with('d');
let is_symlink = parts[0].starts_with('l');
let size = parts[4].parse().unwrap_or(0);
let path = join_ftp(base, &name);
return Some(FileEntry {
path,
name,
is_dir,
is_file: !is_dir && !is_symlink,
is_symlink,
size,
modified: None,
});
}
let parts: Vec<&str> = line.split_whitespace().collect();
if parts.len() >= 4 {
let is_dir = parts[2].eq_ignore_ascii_case("<DIR>");
let (size, name_idx) = if is_dir {
(0u64, 3)
} else {
(parts[2].parse().unwrap_or(0), 3)
};
if name_idx >= parts.len() {
return None;
}
let name = parts[name_idx..].join(" ");
if name == "." || name == ".." {
return None;
}
let path = join_ftp(base, &name);
return Some(FileEntry {
path,
name,
is_dir,
is_file: !is_dir,
is_symlink: false,
size,
modified: None,
});
}
None
}
fn join_ftp(base: &str, name: &str) -> PathBuf {
if base.is_empty() || base == "/" {
PathBuf::from(format!("/{name}"))
} else {
PathBuf::from(base.trim_end_matches('/')).join(name)
}
}
#[async_trait]
impl RemoteFileSystem for FtpClient {
async fn connect(&mut self) -> Result<()> {
let mut stream = FtpStream::connect(self.addr().as_str()).await?;
stream
.login(&self.config.username, &self.config.password)
.await?;
let _ = self.config.passive; self.stream = Some(stream);
Ok(())
}
async fn disconnect(&mut self) -> Result<()> {
if let Some(mut stream) = self.stream.take() {
let _ = stream.quit().await;
}
Ok(())
}
fn is_connected(&self) -> bool {
self.stream.is_some()
}
async fn list(&mut self, path: &str) -> Result<Vec<FileEntry>> {
let stream = self.stream()?;
let pathname = if path.is_empty() || path == "/" {
None
} else {
Some(path)
};
let lines = stream.list(pathname).await?;
Ok(lines
.iter()
.filter_map(|line| parse_list_line(path, line))
.collect())
}
async fn metadata(&mut self, path: &str) -> Result<FileMeta> {
let stream = self.stream()?;
let size = stream.size(path).await?;
let modified = stream
.mdtm(path)
.await?
.map(|dt| SystemTime::from(dt));
if let Some(size) = size {
return Ok(FileMeta {
path: PathBuf::from(path),
is_dir: false,
is_file: true,
is_symlink: false,
size: size as u64,
modified,
});
}
match stream.list(Some(path)).await {
Ok(_) => Ok(FileMeta {
path: PathBuf::from(path),
is_dir: true,
is_file: false,
is_symlink: false,
size: 0,
modified,
}),
Err(err) => Err(err.into()),
}
}
async fn exists(&mut self, path: &str) -> Result<bool> {
match self.metadata(path).await {
Ok(_) => Ok(true),
Err(_) => Ok(false),
}
}
async fn mkdir(&mut self, path: &str) -> Result<()> {
self.stream()?.mkdir(path).await?;
Ok(())
}
async fn mkdir_all(&mut self, path: &str) -> Result<()> {
let path = path.trim_matches('/');
if path.is_empty() {
return Ok(());
}
let mut current = String::new();
for part in path.split('/').filter(|p| !p.is_empty()) {
current = if current.is_empty() {
format!("/{part}")
} else {
format!("{current}/{part}")
};
if !self.exists(¤t).await? {
self.mkdir(¤t).await?;
}
}
Ok(())
}
async fn rmdir(&mut self, path: &str) -> Result<()> {
self.stream()?.rmdir(path).await?;
Ok(())
}
async fn remove_file(&mut self, path: &str) -> Result<()> {
self.stream()?.rm(path).await?;
Ok(())
}
async fn remove(&mut self, path: &str) -> Result<()> {
let meta = self.metadata(path).await?;
if meta.is_dir {
let children = self.list(path).await?;
for child in children {
let child_path = child.path.to_string_lossy().to_string();
Box::pin(self.remove(&child_path)).await?;
}
self.rmdir(path).await
} else {
self.remove_file(path).await
}
}
async fn rename(&mut self, from: &str, to: &str) -> Result<()> {
self.stream()?.rename(from, to).await?;
Ok(())
}
async fn read(&mut self, path: &str) -> Result<Vec<u8>> {
let cursor = self.stream()?.simple_retr(path).await?;
Ok(cursor.into_inner())
}
async fn write(&mut self, path: &str, data: &[u8]) -> Result<()> {
let mut reader = Cursor::new(data.to_vec());
self.stream()?.put(path, &mut reader).await?;
Ok(())
}
async fn download(&mut self, remote: &str, local: &Path) -> Result<()> {
let data = self.read(remote).await?;
tokio::fs::write(local, data).await?;
Ok(())
}
async fn upload(&mut self, local: &Path, remote: &str) -> Result<()> {
let data = tokio::fs::read(local).await?;
self.write(remote, &data).await
}
}
#[allow(dead_code)]
async fn read_data_stream<R: AsyncReadExt + Unpin>(mut reader: R) -> Result<Vec<u8>> {
let mut buf = Vec::new();
reader.read_to_end(&mut buf).await?;
Ok(buf)
}