完成webdav同步功能
This commit is contained in:
@@ -1,14 +1,26 @@
|
||||
// Learn more about Tauri commands at https://tauri.app/develop/calling-rust/
|
||||
use notify::{RecursiveMode, Watcher};
|
||||
use opendal::services::{S3, Webdav};
|
||||
use opendal::{EntryMode, ErrorKind, Operator};
|
||||
use std::collections::{BTreeMap, BTreeSet};
|
||||
use std::future::{Future, IntoFuture};
|
||||
use std::fs;
|
||||
use std::path::PathBuf;
|
||||
use std::hash::{Hash, Hasher};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Mutex;
|
||||
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
|
||||
use tauri::menu::{MenuBuilder, MenuItemBuilder, SubmenuBuilder};
|
||||
use tauri::{Emitter, Manager};
|
||||
use tauri_plugin_dialog::DialogExt;
|
||||
use base64::Engine;
|
||||
use regex::Regex;
|
||||
|
||||
const SYNC_CHECK_TIMEOUT: Duration = Duration::from_secs(15);
|
||||
const SYNC_TEST_STEP_TIMEOUT: Duration = Duration::from_secs(15);
|
||||
const SYNC_REMOTE_LIST_TIMEOUT: Duration = Duration::from_secs(120);
|
||||
const SYNC_REMOTE_FILE_TIMEOUT: Duration = Duration::from_secs(180);
|
||||
const SYNC_TEST_PAYLOAD: &[u8] = b"yurou sync connection probe\n";
|
||||
|
||||
#[derive(Debug, Clone, serde::Serialize)]
|
||||
struct DirEntry {
|
||||
name: String,
|
||||
@@ -455,6 +467,845 @@ fn walk_dir(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct SyncConfig {
|
||||
provider: SyncProvider,
|
||||
webdav: WebdavSyncConfig,
|
||||
s3: S3SyncConfig,
|
||||
auto_sync_on_startup: bool,
|
||||
auto_sync_after_save: bool,
|
||||
last_sync_at: Option<String>,
|
||||
}
|
||||
|
||||
impl Default for SyncConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
provider: SyncProvider::None,
|
||||
webdav: WebdavSyncConfig::default(),
|
||||
s3: S3SyncConfig::default(),
|
||||
auto_sync_on_startup: false,
|
||||
auto_sync_after_save: false,
|
||||
last_sync_at: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
|
||||
#[serde(rename_all = "lowercase")]
|
||||
enum SyncProvider {
|
||||
None,
|
||||
Webdav,
|
||||
S3,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct WebdavSyncConfig {
|
||||
endpoint: String,
|
||||
username: String,
|
||||
password: String,
|
||||
remote_dir: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct S3SyncConfig {
|
||||
endpoint: String,
|
||||
region: String,
|
||||
bucket: String,
|
||||
access_key_id: String,
|
||||
secret_access_key: String,
|
||||
remote_prefix: String,
|
||||
path_style: bool,
|
||||
use_https: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct SyncHistory {
|
||||
provider_key: String,
|
||||
files: BTreeMap<String, SyncHistoryEntry>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct SyncHistoryEntry {
|
||||
local_modified_ms: Option<i128>,
|
||||
remote_modified_ms: Option<i128>,
|
||||
size: u64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
struct SyncFileInfo {
|
||||
modified_ms: Option<i128>,
|
||||
size: u64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, Default, serde::Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct SyncStats {
|
||||
uploaded: usize,
|
||||
downloaded: usize,
|
||||
deleted_local: usize,
|
||||
deleted_remote: usize,
|
||||
skipped: usize,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, serde::Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct SyncResult {
|
||||
message: String,
|
||||
last_sync_at: String,
|
||||
stats: SyncStats,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, serde::Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct SyncConnectionTestResult {
|
||||
message: String,
|
||||
details: Vec<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
enum SyncAction {
|
||||
Upload,
|
||||
Download,
|
||||
DeleteLocal,
|
||||
DeleteRemote,
|
||||
Skip,
|
||||
}
|
||||
|
||||
fn tauri_app_config_dir(app: &tauri::AppHandle) -> Result<PathBuf, String> {
|
||||
app.path()
|
||||
.app_config_dir()
|
||||
.map_err(|e| format!("无法获取配置目录: {}", e))
|
||||
}
|
||||
|
||||
fn sync_config_path(app: &tauri::AppHandle) -> Result<PathBuf, String> {
|
||||
Ok(tauri_app_config_dir(app)?.join("sync-config.json"))
|
||||
}
|
||||
|
||||
fn sync_history_path(app: &tauri::AppHandle, provider_key: &str) -> Result<PathBuf, String> {
|
||||
Ok(tauri_app_config_dir(app)?.join(format!("sync-history-{}.json", stable_hash(provider_key))))
|
||||
}
|
||||
|
||||
fn stable_hash(value: &str) -> u64 {
|
||||
let mut hasher = std::collections::hash_map::DefaultHasher::new();
|
||||
value.hash(&mut hasher);
|
||||
hasher.finish()
|
||||
}
|
||||
|
||||
fn load_sync_config_from_disk(app: &tauri::AppHandle) -> Result<SyncConfig, String> {
|
||||
let path = sync_config_path(app)?;
|
||||
if !path.exists() {
|
||||
return Ok(SyncConfig::default());
|
||||
}
|
||||
|
||||
let raw = fs::read_to_string(&path).map_err(|e| format!("无法读取同步配置: {}", e))?;
|
||||
serde_json::from_str(&raw).map_err(|e| format!("同步配置格式无效: {}", e))
|
||||
}
|
||||
|
||||
fn save_sync_config_to_disk(app: &tauri::AppHandle, config: &SyncConfig) -> Result<(), String> {
|
||||
let path = sync_config_path(app)?;
|
||||
if let Some(parent) = path.parent() {
|
||||
fs::create_dir_all(parent).map_err(|e| format!("无法创建配置目录: {}", e))?;
|
||||
}
|
||||
|
||||
let raw = serde_json::to_string_pretty(config).map_err(|e| format!("无法序列化同步配置: {}", e))?;
|
||||
fs::write(&path, raw).map_err(|e| format!("无法保存同步配置: {}", e))
|
||||
}
|
||||
|
||||
fn load_sync_history(app: &tauri::AppHandle, provider_key: &str) -> Result<SyncHistory, String> {
|
||||
let path = sync_history_path(app, provider_key)?;
|
||||
if !path.exists() {
|
||||
return Ok(SyncHistory {
|
||||
provider_key: provider_key.to_string(),
|
||||
files: BTreeMap::new(),
|
||||
});
|
||||
}
|
||||
|
||||
let raw = fs::read_to_string(&path).map_err(|e| format!("无法读取同步历史: {}", e))?;
|
||||
let mut history: SyncHistory =
|
||||
serde_json::from_str(&raw).map_err(|e| format!("同步历史格式无效: {}", e))?;
|
||||
history.provider_key = provider_key.to_string();
|
||||
Ok(history)
|
||||
}
|
||||
|
||||
fn save_sync_history(app: &tauri::AppHandle, history: &SyncHistory) -> Result<(), String> {
|
||||
let path = sync_history_path(app, &history.provider_key)?;
|
||||
if let Some(parent) = path.parent() {
|
||||
fs::create_dir_all(parent).map_err(|e| format!("无法创建同步历史目录: {}", e))?;
|
||||
}
|
||||
|
||||
let raw = serde_json::to_string_pretty(history).map_err(|e| format!("无法序列化同步历史: {}", e))?;
|
||||
fs::write(&path, raw).map_err(|e| format!("无法保存同步历史: {}", e))
|
||||
}
|
||||
|
||||
fn normalize_remote_root(path: &str) -> String {
|
||||
let trimmed = path.trim().replace('\\', "/");
|
||||
let trimmed = trimmed.trim_matches('/');
|
||||
if trimmed.is_empty() {
|
||||
"/".to_string()
|
||||
} else {
|
||||
format!("/{}", trimmed)
|
||||
}
|
||||
}
|
||||
|
||||
fn normalize_s3_prefix(prefix: &str) -> String {
|
||||
prefix
|
||||
.trim()
|
||||
.replace('\\', "/")
|
||||
.trim_matches('/')
|
||||
.to_string()
|
||||
}
|
||||
|
||||
fn normalize_remote_file_path(path: &str) -> String {
|
||||
path.trim_start_matches('/').replace('\\', "/")
|
||||
}
|
||||
|
||||
fn normalize_relative_path(path: &Path) -> String {
|
||||
path.components()
|
||||
.filter_map(|component| match component {
|
||||
std::path::Component::Normal(part) => Some(part.to_string_lossy().to_string()),
|
||||
_ => None,
|
||||
})
|
||||
.collect::<Vec<_>>()
|
||||
.join("/")
|
||||
}
|
||||
|
||||
fn provider_key(config: &SyncConfig) -> Result<String, String> {
|
||||
match config.provider {
|
||||
SyncProvider::None => Err("请先选择同步方式".to_string()),
|
||||
SyncProvider::Webdav => {
|
||||
let endpoint = config.webdav.endpoint.trim();
|
||||
if endpoint.is_empty() {
|
||||
return Err("请填写 WebDAV 服务器地址".to_string());
|
||||
}
|
||||
Ok(format!(
|
||||
"webdav:{}:{}",
|
||||
endpoint.trim_end_matches('/'),
|
||||
normalize_remote_root(&config.webdav.remote_dir)
|
||||
))
|
||||
}
|
||||
SyncProvider::S3 => {
|
||||
let bucket = config.s3.bucket.trim();
|
||||
if bucket.is_empty() {
|
||||
return Err("请填写 S3 Bucket".to_string());
|
||||
}
|
||||
Ok(format!(
|
||||
"s3:{}:{}:{}",
|
||||
config.s3.endpoint.trim().trim_end_matches('/'),
|
||||
bucket,
|
||||
normalize_s3_prefix(&config.s3.remote_prefix)
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn build_sync_operator(config: &SyncConfig) -> Result<Operator, String> {
|
||||
match config.provider {
|
||||
SyncProvider::None => Err("请先选择同步方式".to_string()),
|
||||
SyncProvider::Webdav => {
|
||||
let endpoint = config.webdav.endpoint.trim();
|
||||
if endpoint.is_empty() {
|
||||
return Err("请填写 WebDAV 服务器地址".to_string());
|
||||
}
|
||||
|
||||
let mut builder = Webdav::default()
|
||||
.endpoint(endpoint)
|
||||
.root(&normalize_remote_root(&config.webdav.remote_dir));
|
||||
|
||||
if !config.webdav.username.trim().is_empty() {
|
||||
builder = builder.username(config.webdav.username.trim());
|
||||
}
|
||||
if !config.webdav.password.is_empty() {
|
||||
builder = builder.password(&config.webdav.password);
|
||||
}
|
||||
|
||||
Operator::new(builder)
|
||||
.map(|builder| builder.finish())
|
||||
.map_err(|e| format!("无法初始化 WebDAV 同步: {}", e))
|
||||
}
|
||||
SyncProvider::S3 => {
|
||||
let bucket = config.s3.bucket.trim();
|
||||
if bucket.is_empty() {
|
||||
return Err("请填写 S3 Bucket".to_string());
|
||||
}
|
||||
|
||||
let mut endpoint = config.s3.endpoint.trim().trim_end_matches('/').to_string();
|
||||
if !endpoint.is_empty() && !endpoint.contains("://") {
|
||||
endpoint = if config.s3.use_https {
|
||||
format!("https://{}", endpoint)
|
||||
} else {
|
||||
format!("http://{}", endpoint)
|
||||
};
|
||||
}
|
||||
|
||||
let mut builder = S3::default()
|
||||
.bucket(bucket)
|
||||
.root(&normalize_s3_prefix(&config.s3.remote_prefix))
|
||||
.disable_config_load();
|
||||
|
||||
if !endpoint.is_empty() {
|
||||
builder = builder.endpoint(&endpoint);
|
||||
}
|
||||
if !config.s3.region.trim().is_empty() {
|
||||
builder = builder.region(config.s3.region.trim());
|
||||
}
|
||||
if !config.s3.access_key_id.trim().is_empty() {
|
||||
builder = builder.access_key_id(config.s3.access_key_id.trim());
|
||||
}
|
||||
if !config.s3.secret_access_key.is_empty() {
|
||||
builder = builder.secret_access_key(&config.s3.secret_access_key);
|
||||
}
|
||||
if !config.s3.path_style {
|
||||
builder = builder.enable_virtual_host_style();
|
||||
}
|
||||
|
||||
Operator::new(builder)
|
||||
.map(|builder| builder.finish())
|
||||
.map_err(|e| format!("无法初始化 S3 同步: {}", e))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn opendal_timeout<T, F>(step: &str, timeout: Duration, future: F) -> Result<T, String>
|
||||
where
|
||||
F: Future<Output = Result<T, opendal::Error>>,
|
||||
{
|
||||
match tokio::time::timeout(timeout, future).await {
|
||||
Ok(Ok(value)) => Ok(value),
|
||||
Ok(Err(error)) => Err(format!("{}失败: {}", step, describe_opendal_error(&error))),
|
||||
Err(_) => Err(format!("{}超时({} 秒)", step, timeout.as_secs())),
|
||||
}
|
||||
}
|
||||
|
||||
async fn test_step<T, F>(
|
||||
details: &mut Vec<String>,
|
||||
step: &str,
|
||||
success: &str,
|
||||
future: F,
|
||||
) -> Result<T, String>
|
||||
where
|
||||
F: Future<Output = Result<T, opendal::Error>>,
|
||||
{
|
||||
let start = Instant::now();
|
||||
match tokio::time::timeout(SYNC_TEST_STEP_TIMEOUT, future).await {
|
||||
Ok(Ok(value)) => {
|
||||
details.push(format!(
|
||||
"{}: {},耗时 {}ms",
|
||||
step,
|
||||
success,
|
||||
start.elapsed().as_millis()
|
||||
));
|
||||
Ok(value)
|
||||
}
|
||||
Ok(Err(error)) => Err(format_sync_test_failure(
|
||||
step,
|
||||
describe_opendal_error(&error),
|
||||
details,
|
||||
)),
|
||||
Err(_) => Err(format_sync_test_failure(
|
||||
step,
|
||||
format!("请求超时({} 秒),请检查服务器地址、网络或远端目录响应速度", SYNC_TEST_STEP_TIMEOUT.as_secs()),
|
||||
details,
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
fn describe_opendal_error(error: &opendal::Error) -> String {
|
||||
let hint = match error.kind() {
|
||||
ErrorKind::ConfigInvalid => "配置无效,请检查服务器地址、Bucket、Region 或远端路径",
|
||||
ErrorKind::PermissionDenied => "权限被拒绝,通常表示账号/密码/Access Key 不正确,或该账号没有远端目录权限",
|
||||
ErrorKind::NotFound => "远端路径不存在,请检查 WebDAV 远端目录、S3 Bucket 或前缀",
|
||||
ErrorKind::Unsupported => "服务不支持当前操作,请确认 WebDAV/S3 兼容能力",
|
||||
ErrorKind::RateLimited => "服务限流,请稍后重试",
|
||||
ErrorKind::IsADirectory => "目标是目录,不是文件",
|
||||
ErrorKind::NotADirectory => "目标不是目录",
|
||||
ErrorKind::AlreadyExists => "目标已存在",
|
||||
ErrorKind::ConditionNotMatch => "服务端条件校验未通过",
|
||||
ErrorKind::RangeNotSatisfied => "读取范围无效",
|
||||
ErrorKind::IsSameFile => "源和目标是同一个文件",
|
||||
ErrorKind::Unexpected => "服务返回了非预期错误,请检查原始错误信息",
|
||||
_ => "服务返回了未知错误",
|
||||
};
|
||||
|
||||
format!("{};原始错误: {}", hint, error)
|
||||
}
|
||||
|
||||
fn format_sync_test_failure(step: &str, reason: String, details: &[String]) -> String {
|
||||
let mut lines = vec![format!("连接测试失败,失败步骤:{}", step), reason];
|
||||
if !details.is_empty() {
|
||||
lines.push("已通过步骤:".to_string());
|
||||
lines.extend(details.iter().map(|detail| format!("- {}", detail)));
|
||||
}
|
||||
lines.join("\n")
|
||||
}
|
||||
|
||||
fn sync_target_label(config: &SyncConfig) -> String {
|
||||
match config.provider {
|
||||
SyncProvider::None => "未选择同步方式".to_string(),
|
||||
SyncProvider::Webdav => format!(
|
||||
"WebDAV {}{}",
|
||||
config.webdav.endpoint.trim(),
|
||||
normalize_remote_root(&config.webdav.remote_dir)
|
||||
),
|
||||
SyncProvider::S3 => format!(
|
||||
"S3 bucket={} prefix={}",
|
||||
config.s3.bucket.trim(),
|
||||
normalize_s3_prefix(&config.s3.remote_prefix)
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
fn sync_probe_file_path() -> String {
|
||||
let millis = SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|duration| duration.as_millis())
|
||||
.unwrap_or_default();
|
||||
format!("_yurou-sync-probe-{}-{}.tmp", std::process::id(), millis)
|
||||
}
|
||||
|
||||
fn remote_parent_dirs(relative: &str) -> Vec<String> {
|
||||
let parts: Vec<&str> = relative.split('/').filter(|part| !part.is_empty()).collect();
|
||||
if parts.len() <= 1 {
|
||||
return Vec::new();
|
||||
}
|
||||
|
||||
let mut dirs = Vec::new();
|
||||
let mut current = String::new();
|
||||
for part in parts.iter().take(parts.len() - 1) {
|
||||
current.push_str(part);
|
||||
current.push('/');
|
||||
dirs.push(current.clone());
|
||||
}
|
||||
dirs
|
||||
}
|
||||
|
||||
async fn ensure_remote_parent_dirs(op: &Operator, relative: &str) -> Result<(), String> {
|
||||
for dir in remote_parent_dirs(relative) {
|
||||
match tokio::time::timeout(SYNC_REMOTE_FILE_TIMEOUT, op.create_dir(&dir)).await {
|
||||
Ok(Ok(())) => {}
|
||||
Ok(Err(error)) if error.kind() == ErrorKind::AlreadyExists => {}
|
||||
Ok(Err(error)) => {
|
||||
return Err(format!(
|
||||
"创建远端目录 {}失败: {}",
|
||||
dir,
|
||||
describe_opendal_error(&error)
|
||||
));
|
||||
}
|
||||
Err(_) => {
|
||||
return Err(format!(
|
||||
"创建远端目录 {}超时({} 秒)",
|
||||
dir,
|
||||
SYNC_REMOTE_FILE_TIMEOUT.as_secs()
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn run_sync_connection_test(config: &SyncConfig) -> Result<SyncConnectionTestResult, String> {
|
||||
let mut details = Vec::new();
|
||||
let op = build_sync_operator(config)?;
|
||||
details.push(format!("配置解析通过,目标:{}", sync_target_label(config)));
|
||||
|
||||
test_step(&mut details, "服务检查", "服务器可达", op.check()).await?;
|
||||
test_step(
|
||||
&mut details,
|
||||
"目录读取",
|
||||
"账号/密码有效,远端目录可读取",
|
||||
op.list_with("/").limit(1).into_future(),
|
||||
)
|
||||
.await?;
|
||||
|
||||
let probe_path = sync_probe_file_path();
|
||||
test_step(
|
||||
&mut details,
|
||||
"写入探针",
|
||||
"远端目录可写",
|
||||
op.write(&probe_path, SYNC_TEST_PAYLOAD.to_vec()),
|
||||
)
|
||||
.await?;
|
||||
|
||||
let read_result = test_step(
|
||||
&mut details,
|
||||
"读取探针",
|
||||
"写入内容可读回",
|
||||
op.read(&probe_path),
|
||||
)
|
||||
.await;
|
||||
let read_bytes = match read_result {
|
||||
Ok(bytes) => bytes.to_vec(),
|
||||
Err(error) => {
|
||||
let _ = opendal_timeout("清理测试文件", SYNC_TEST_STEP_TIMEOUT, op.delete(&probe_path)).await;
|
||||
return Err(error);
|
||||
}
|
||||
};
|
||||
|
||||
if read_bytes != SYNC_TEST_PAYLOAD {
|
||||
let _ = opendal_timeout("清理测试文件", SYNC_TEST_STEP_TIMEOUT, op.delete(&probe_path)).await;
|
||||
return Err(format_sync_test_failure(
|
||||
"读取探针",
|
||||
"读回内容与写入内容不一致,请检查 WebDAV/S3 服务是否有缓存或代理改写".to_string(),
|
||||
&details,
|
||||
));
|
||||
}
|
||||
|
||||
test_step(
|
||||
&mut details,
|
||||
"删除探针",
|
||||
"远端目录可删除",
|
||||
op.delete(&probe_path),
|
||||
)
|
||||
.await?;
|
||||
|
||||
Ok(SyncConnectionTestResult {
|
||||
message: "连接可用,账号/密码正确,远端目录具备读取、写入和删除权限".to_string(),
|
||||
details,
|
||||
})
|
||||
}
|
||||
|
||||
fn system_time_ms(time: SystemTime) -> Option<i128> {
|
||||
match time.duration_since(UNIX_EPOCH) {
|
||||
Ok(duration) => Some(duration.as_millis() as i128),
|
||||
Err(error) => Some(-(error.duration().as_millis() as i128)),
|
||||
}
|
||||
}
|
||||
|
||||
fn now_string() -> String {
|
||||
opendal::raw::Timestamp::now().to_string()
|
||||
}
|
||||
|
||||
fn remote_modified_ms(meta: &opendal::Metadata) -> Option<i128> {
|
||||
meta.last_modified()
|
||||
.and_then(|value| {
|
||||
let system_time: SystemTime = value.into();
|
||||
system_time_ms(system_time)
|
||||
})
|
||||
}
|
||||
|
||||
fn scan_local_files(root: &Path) -> Result<BTreeMap<String, SyncFileInfo>, String> {
|
||||
if !root.is_dir() {
|
||||
return Err("工作目录不存在或不是文件夹".to_string());
|
||||
}
|
||||
|
||||
let mut files = BTreeMap::new();
|
||||
scan_local_files_inner(root, root, &mut files)?;
|
||||
Ok(files)
|
||||
}
|
||||
|
||||
fn scan_local_files_inner(
|
||||
root: &Path,
|
||||
current: &Path,
|
||||
files: &mut BTreeMap<String, SyncFileInfo>,
|
||||
) -> Result<(), String> {
|
||||
for entry in fs::read_dir(current).map_err(|e| format!("无法读取工作目录: {}", e))? {
|
||||
let entry = entry.map_err(|e| format!("读取工作目录条目失败: {}", e))?;
|
||||
let path = entry.path();
|
||||
let file_type = entry
|
||||
.file_type()
|
||||
.map_err(|e| format!("无法读取文件类型: {}", e))?;
|
||||
|
||||
if file_type.is_dir() {
|
||||
scan_local_files_inner(root, &path, files)?;
|
||||
} else if file_type.is_file() {
|
||||
let metadata = entry
|
||||
.metadata()
|
||||
.map_err(|e| format!("无法读取文件元数据: {}", e))?;
|
||||
let relative = path
|
||||
.strip_prefix(root)
|
||||
.map(normalize_relative_path)
|
||||
.map_err(|e| format!("无法计算相对路径: {}", e))?;
|
||||
files.insert(
|
||||
relative.clone(),
|
||||
SyncFileInfo {
|
||||
modified_ms: metadata.modified().ok().and_then(system_time_ms),
|
||||
size: metadata.len(),
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn scan_remote_files(op: &Operator) -> Result<BTreeMap<String, SyncFileInfo>, String> {
|
||||
let entries = opendal_timeout(
|
||||
"读取远端文件列表",
|
||||
SYNC_REMOTE_LIST_TIMEOUT,
|
||||
op.list_with("/").recursive(true).into_future(),
|
||||
)
|
||||
.await?;
|
||||
|
||||
let mut files = BTreeMap::new();
|
||||
for entry in entries {
|
||||
let (path, listed_meta) = entry.into_parts();
|
||||
if path == "/" || path.ends_with('/') || listed_meta.mode() == EntryMode::DIR {
|
||||
continue;
|
||||
}
|
||||
|
||||
let meta = match tokio::time::timeout(SYNC_REMOTE_FILE_TIMEOUT, op.stat(&path)).await {
|
||||
Ok(Ok(meta)) => meta,
|
||||
Ok(Err(error)) if error.kind() == ErrorKind::NotFound => continue,
|
||||
Ok(Err(error)) => {
|
||||
return Err(format!(
|
||||
"读取远端文件元数据 {}失败: {}",
|
||||
path,
|
||||
describe_opendal_error(&error)
|
||||
));
|
||||
}
|
||||
Err(_) => {
|
||||
return Err(format!(
|
||||
"读取远端文件元数据 {}超时({} 秒)",
|
||||
path,
|
||||
SYNC_REMOTE_FILE_TIMEOUT.as_secs()
|
||||
));
|
||||
}
|
||||
};
|
||||
|
||||
if !meta.is_file() {
|
||||
continue;
|
||||
}
|
||||
|
||||
let relative = normalize_remote_file_path(&path);
|
||||
files.insert(
|
||||
relative.clone(),
|
||||
SyncFileInfo {
|
||||
modified_ms: remote_modified_ms(&meta),
|
||||
size: meta.content_length(),
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
Ok(files)
|
||||
}
|
||||
|
||||
fn choose_sync_action(
|
||||
local: Option<&SyncFileInfo>,
|
||||
remote: Option<&SyncFileInfo>,
|
||||
history: Option<&SyncHistoryEntry>,
|
||||
) -> SyncAction {
|
||||
match (local, remote) {
|
||||
(Some(local), Some(remote)) => {
|
||||
if let Some(history) = history {
|
||||
if history.local_modified_ms == local.modified_ms
|
||||
&& history.remote_modified_ms == remote.modified_ms
|
||||
&& history.size == local.size
|
||||
&& history.size == remote.size
|
||||
{
|
||||
return SyncAction::Skip;
|
||||
}
|
||||
}
|
||||
|
||||
if local.size == remote.size
|
||||
&& local.modified_ms.is_some()
|
||||
&& remote.modified_ms.is_some()
|
||||
&& local.modified_ms == remote.modified_ms
|
||||
{
|
||||
return SyncAction::Skip;
|
||||
}
|
||||
|
||||
match (local.modified_ms, remote.modified_ms) {
|
||||
(Some(local_ms), Some(remote_ms)) => {
|
||||
if local_ms >= remote_ms {
|
||||
SyncAction::Upload
|
||||
} else {
|
||||
SyncAction::Download
|
||||
}
|
||||
}
|
||||
(Some(_), None) => SyncAction::Upload,
|
||||
(None, Some(_)) => SyncAction::Download,
|
||||
(None, None) => {
|
||||
if local.size == remote.size {
|
||||
SyncAction::Skip
|
||||
} else {
|
||||
SyncAction::Upload
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
(Some(local), None) => {
|
||||
if let Some(history) = history {
|
||||
if history.local_modified_ms == local.modified_ms && history.size == local.size {
|
||||
SyncAction::DeleteLocal
|
||||
} else {
|
||||
SyncAction::Upload
|
||||
}
|
||||
} else {
|
||||
SyncAction::Upload
|
||||
}
|
||||
}
|
||||
(None, Some(remote)) => {
|
||||
if let Some(history) = history {
|
||||
if history.remote_modified_ms == remote.modified_ms && history.size == remote.size {
|
||||
SyncAction::DeleteRemote
|
||||
} else {
|
||||
SyncAction::Download
|
||||
}
|
||||
} else {
|
||||
SyncAction::Download
|
||||
}
|
||||
}
|
||||
(None, None) => SyncAction::Skip,
|
||||
}
|
||||
}
|
||||
|
||||
fn local_path_for(root: &Path, relative: &str) -> PathBuf {
|
||||
relative
|
||||
.split('/')
|
||||
.filter(|part| !part.is_empty())
|
||||
.fold(root.to_path_buf(), |path, part| path.join(part))
|
||||
}
|
||||
|
||||
async fn sync_workspace(
|
||||
app: &tauri::AppHandle,
|
||||
workspace: &Path,
|
||||
config: &SyncConfig,
|
||||
) -> Result<SyncResult, String> {
|
||||
let op = build_sync_operator(config)?;
|
||||
opendal_timeout("同步连接检查", SYNC_CHECK_TIMEOUT, op.check()).await?;
|
||||
|
||||
let key = provider_key(config)?;
|
||||
let local_files = scan_local_files(workspace)?;
|
||||
let remote_files = scan_remote_files(&op).await?;
|
||||
let mut history = load_sync_history(app, &key)?;
|
||||
let mut stats = SyncStats::default();
|
||||
|
||||
let all_paths: BTreeSet<String> = local_files
|
||||
.keys()
|
||||
.chain(remote_files.keys())
|
||||
.chain(history.files.keys())
|
||||
.cloned()
|
||||
.collect();
|
||||
|
||||
for relative in all_paths {
|
||||
let local = local_files.get(&relative);
|
||||
let remote = remote_files.get(&relative);
|
||||
let prior = history.files.get(&relative);
|
||||
|
||||
match choose_sync_action(local, remote, prior) {
|
||||
SyncAction::Upload => {
|
||||
let local_path = local_path_for(workspace, &relative);
|
||||
let bytes = fs::read(&local_path)
|
||||
.map_err(|e| format!("无法读取本地文件 {}: {}", local_path.display(), e))?;
|
||||
ensure_remote_parent_dirs(&op, &relative).await?;
|
||||
opendal_timeout(
|
||||
&format!("上传 {}", relative),
|
||||
SYNC_REMOTE_FILE_TIMEOUT,
|
||||
op.write(&relative, bytes),
|
||||
)
|
||||
.await?;
|
||||
stats.uploaded += 1;
|
||||
}
|
||||
SyncAction::Download => {
|
||||
let local_path = local_path_for(workspace, &relative);
|
||||
if let Some(parent) = local_path.parent() {
|
||||
fs::create_dir_all(parent)
|
||||
.map_err(|e| format!("无法创建本地目录 {}: {}", parent.display(), e))?;
|
||||
}
|
||||
let bytes = opendal_timeout(
|
||||
&format!("下载 {}", relative),
|
||||
SYNC_REMOTE_FILE_TIMEOUT,
|
||||
op.read(&relative),
|
||||
)
|
||||
.await?
|
||||
.to_vec();
|
||||
fs::write(&local_path, bytes)
|
||||
.map_err(|e| format!("无法写入本地文件 {}: {}", local_path.display(), e))?;
|
||||
stats.downloaded += 1;
|
||||
}
|
||||
SyncAction::DeleteLocal => {
|
||||
let local_path = local_path_for(workspace, &relative);
|
||||
if local_path.is_file() {
|
||||
fs::remove_file(&local_path)
|
||||
.map_err(|e| format!("无法删除本地文件 {}: {}", local_path.display(), e))?;
|
||||
}
|
||||
history.files.remove(&relative);
|
||||
stats.deleted_local += 1;
|
||||
continue;
|
||||
}
|
||||
SyncAction::DeleteRemote => {
|
||||
opendal_timeout(
|
||||
&format!("删除远端文件 {}", relative),
|
||||
SYNC_REMOTE_FILE_TIMEOUT,
|
||||
op.delete(&relative),
|
||||
)
|
||||
.await?;
|
||||
history.files.remove(&relative);
|
||||
stats.deleted_remote += 1;
|
||||
continue;
|
||||
}
|
||||
SyncAction::Skip => {
|
||||
stats.skipped += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let refreshed_local = scan_local_files(workspace)?;
|
||||
let refreshed_remote = scan_remote_files(&op).await?;
|
||||
history.files.clear();
|
||||
|
||||
let synced_paths: BTreeSet<String> = refreshed_local
|
||||
.keys()
|
||||
.chain(refreshed_remote.keys())
|
||||
.cloned()
|
||||
.collect();
|
||||
|
||||
for relative in synced_paths {
|
||||
let local = refreshed_local.get(&relative);
|
||||
let remote = refreshed_remote.get(&relative);
|
||||
let size = local
|
||||
.map(|item| item.size)
|
||||
.or_else(|| remote.map(|item| item.size))
|
||||
.unwrap_or(0);
|
||||
history.files.insert(
|
||||
relative,
|
||||
SyncHistoryEntry {
|
||||
local_modified_ms: local.and_then(|item| item.modified_ms),
|
||||
remote_modified_ms: remote.and_then(|item| item.modified_ms),
|
||||
size,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
save_sync_history(app, &history)?;
|
||||
|
||||
let last_sync_at = now_string();
|
||||
let mut saved_config = config.clone();
|
||||
saved_config.last_sync_at = Some(last_sync_at.clone());
|
||||
save_sync_config_to_disk(app, &saved_config)?;
|
||||
|
||||
Ok(SyncResult {
|
||||
message: "同步完成".to_string(),
|
||||
last_sync_at,
|
||||
stats,
|
||||
})
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
fn load_sync_config(app: tauri::AppHandle) -> Result<SyncConfig, String> {
|
||||
load_sync_config_from_disk(&app)
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
fn save_sync_config(app: tauri::AppHandle, config: SyncConfig) -> Result<(), String> {
|
||||
save_sync_config_to_disk(&app, &config)
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
async fn test_sync_connection(config: SyncConfig) -> Result<SyncConnectionTestResult, String> {
|
||||
run_sync_connection_test(&config).await
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
async fn sync_now(app: tauri::AppHandle, workspace: String, config: SyncConfig) -> Result<SyncResult, String> {
|
||||
let workspace = PathBuf::from(workspace);
|
||||
sync_workspace(&app, &workspace, &config).await
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
fn greet(name: &str) -> String {
|
||||
format!("Hello, {}! You've been greeted from Rust!", name)
|
||||
@@ -480,7 +1331,11 @@ pub fn run() {
|
||||
delete_entry,
|
||||
rename_entry,
|
||||
save_media_file,
|
||||
search_files
|
||||
search_files,
|
||||
load_sync_config,
|
||||
save_sync_config,
|
||||
test_sync_connection,
|
||||
sync_now
|
||||
])
|
||||
.setup(|app| {
|
||||
// ---- File menu ----
|
||||
|
||||
Reference in New Issue
Block a user