1use crate::config::AccountConfig;
2use crate::error::{ProviderError, SyncError};
3use crate::providers::{DownloadResult, FileInfo, StorageProvider, UploadResult};
4use async_trait::async_trait;
5use base64::Engine;
6use reqwest::{Client, Method, StatusCode, Url};
7use std::path::Path;
8use std::time::{Duration, SystemTime, UNIX_EPOCH};
9use tracing::{debug, error, info, instrument, warn};
10
11pub struct WebDavProvider {
13 client: Client,
14 base_url: String,
15 path_prefix: String,
16 username: String,
17 password: String,
18}
19
20impl WebDavProvider {
21 #[instrument(skip(config), fields(account_id = %config.id, account_name = %config.name))]
23 pub async fn new(config: &AccountConfig) -> Result<Self, ProviderError> {
24 info!("初始化 WebDAV Provider");
25
26 let url = config.credentials.get("url").ok_or_else(|| {
27 error!("配置缺少 URL");
28 ProviderError::MissingCredentials("url".to_string())
29 })?;
30
31 let username = config.credentials.get("username").ok_or_else(|| {
32 error!("配置缺少用户名");
33 ProviderError::MissingCredentials("username".to_string())
34 })?;
35
36 let password = config.credentials.get("password").ok_or_else(|| {
37 error!("配置缺少密码");
38 ProviderError::MissingCredentials("password".to_string())
39 })?;
40
41 debug!(url = %url, username = %username, "解析 WebDAV 凭证");
42
43 let client = Client::builder()
44 .timeout(Duration::from_secs(30))
45 .build()
46 .map_err(|e| {
47 error!(error = %e, "创建 HTTP 客户端失败");
48 ProviderError::ConnectionFailed(e.to_string())
49 })?;
50
51 let parsed_url = Url::parse(url).map_err(|e| {
52 error!(error = %e, "URL 解析失败");
53 ProviderError::ConnectionFailed(format!("Invalid URL: {}", e))
54 })?;
55
56 let path_prefix = urlencoding::decode(parsed_url.path())
57 .unwrap_or(std::borrow::Cow::Borrowed(parsed_url.path()))
58 .trim_end_matches('/')
59 .to_string();
60
61 info!(base_url = %url, path_prefix = %path_prefix, "WebDAV Provider 初始化成功");
62
63 Ok(Self {
64 client,
65 base_url: url.trim_end_matches('/').to_string(),
66 path_prefix,
67 username: username.clone(),
68 password: password.clone(),
69 })
70 }
71
72 fn get_full_url(&self, path: &str) -> String {
74 let path = path.trim_start_matches('/');
75
76 let encoded_path: Vec<String> = path
78 .split('/')
79 .map(|component| urlencoding::encode(component).to_string())
80 .collect();
81
82 let encoded_path_str = encoded_path.join("/");
83
84 let url = format!("{}/{}", self.base_url, encoded_path_str);
85 debug!(path = %path, url = %url, "构建完整 URL");
86 url
87 }
88
89 fn create_auth_header(&self) -> String {
91 let credentials = format!("{}:{}", self.username, self.password);
92 let encoded = base64::engine::general_purpose::STANDARD.encode(credentials);
93 format!("Basic {}", encoded)
94 }
95
96 #[instrument(skip(self, xml), fields(base_path = %base_path))]
98 fn parse_propfind_response(
99 &self,
100 xml: &str,
101 base_path: &str,
102 ) -> Result<Vec<FileInfo>, SyncError> {
103 debug!("开始解析 PROPFIND 响应");
104 use quick_xml::events::Event;
105 use quick_xml::reader::Reader;
106
107 let mut files = Vec::new();
108 let mut reader = Reader::from_str(xml);
109 reader.config_mut().trim_text(true);
110
111 let mut buf = Vec::new();
112
113 let mut current_path: Option<String> = None;
114 let mut current_size: u64 = 0;
115 let mut is_collection = false;
116
117 let mut in_response = false;
119 let mut in_href = false;
120 let mut in_prop = false;
121 let mut in_getcontentlength = false;
122 let mut in_resourcetype = false;
123 let mut in_collection = false;
124
125 loop {
126 match reader.read_event_into(&mut buf) {
127 Ok(Event::Start(ref e)) => {
128 let name = e.name();
129 let name_str = String::from_utf8_lossy(name.as_ref()).to_lowercase();
130
131 if name_str.ends_with("response") {
132 in_response = true;
133 current_path = None;
134 current_size = 0;
135 is_collection = false;
136 } else if in_response {
137 if name_str.ends_with("href") {
138 in_href = true;
139 } else if name_str.ends_with("prop") {
140 in_prop = true;
141 } else if in_prop {
142 if name_str.ends_with("getcontentlength") {
143 in_getcontentlength = true;
144 } else if name_str.ends_with("resourcetype") {
145 in_resourcetype = true;
146 } else if in_resourcetype && name_str.ends_with("collection") {
147 is_collection = true;
148 }
149 }
150 }
151 }
152 Ok(Event::Empty(ref e)) => {
153 let name = e.name();
154 let name_str = String::from_utf8_lossy(name.as_ref()).to_lowercase();
155 if in_resourcetype && name_str.ends_with("collection") {
156 is_collection = true;
157 }
158 }
159 Ok(Event::Text(e)) => {
160 if in_href {
161 let href = String::from_utf8_lossy(e.as_ref()).to_string();
164
165 let decoded_href =
167 urlencoding::decode(&href).unwrap_or(std::borrow::Cow::Borrowed(&href));
168 let mut path = decoded_href.to_string();
169
170 if path.starts_with(&self.base_url) {
171 path = path.trim_start_matches(&self.base_url).to_string();
172 } else if path.starts_with(&self.path_prefix) {
173 path = path.trim_start_matches(&self.path_prefix).to_string();
174 }
175
176 if !path.starts_with('/') && !path.is_empty() {
178 path = format!("/{}", path);
179 }
180
181 if path.is_empty() {
183 path = "/".to_string();
184 }
185
186 current_path = Some(path);
187 } else if in_getcontentlength {
188 let size_str = String::from_utf8_lossy(e.as_ref()).to_string();
189 if let Ok(size) = size_str.parse::<u64>() {
190 current_size = size;
191 }
192 }
193 }
194 Ok(Event::End(ref e)) => {
195 let name = e.name();
196 let name_str = String::from_utf8_lossy(name.as_ref()).to_lowercase();
197
198 if name_str.ends_with("response") {
199 if let Some(path) = current_path.take() {
200 let norm_path = path.trim_end_matches('/');
203 let norm_base = base_path.trim_end_matches('/');
204
205 debug!(path = %path, norm_path = %norm_path, base = %base_path, norm_base = %norm_base, "Checking if path is base path");
207
208 if norm_path != norm_base && !path.is_empty() {
209 files.push(FileInfo {
210 path, size: current_size,
212 modified: SystemTime::now()
213 .duration_since(UNIX_EPOCH)
214 .unwrap()
215 .as_secs()
216 as i64,
217 hash: None,
218 is_dir: is_collection,
219 });
220 }
221 }
222 in_response = false;
223 } else if name_str.ends_with("href") {
224 in_href = false;
225 } else if name_str.ends_with("prop") {
226 in_prop = false;
227 } else if name_str.ends_with("getcontentlength") {
228 in_getcontentlength = false;
229 } else if name_str.ends_with("resourcetype") {
230 in_resourcetype = false;
231 }
232 }
233 Ok(Event::Eof) => break,
234 Err(e) => {
235 error!("Error parsing XML: {:?}", e);
236 break;
237 }
238 _ => {}
239 }
240 buf.clear();
241 }
242
243 info!(
244 count = files.len(),
245 "解析完成,共 {} 个文件/目录",
246 files.len()
247 );
248 Ok(files)
249 }
250}
251
252#[async_trait]
253impl StorageProvider for WebDavProvider {
254 async fn list(&self, path: &str) -> Result<Vec<FileInfo>, SyncError> {
256 let url = self.get_full_url(path);
257
258 let response = self
259 .client
260 .request(Method::from_bytes(b"PROPFIND").unwrap(), &url)
261 .header("Authorization", self.create_auth_header())
262 .header("Depth", "1")
263 .header("Content-Type", "application/xml")
264 .body(
265 r#"<?xml version="1.0" encoding="utf-8"?>
266 <d:propfind xmlns:d="DAV:">
267 <d:prop>
268 <d:displayname/>
269 <d:getcontentlength/>
270 <d:getlastmodified/>
271 <d:resourcetype/>
272 </d:prop>
273 </d:propfind>"#,
274 )
275 .send()
276 .await
277 .map_err(|e| SyncError::Network(e))?;
278
279 if !response.status().is_success() {
280 return Err(SyncError::Provider(ProviderError::ApiError(format!(
281 "PROPFIND failed: {}",
282 response.status()
283 ))));
284 }
285
286 let body = response.text().await.map_err(|e| SyncError::Network(e))?;
287
288 debug!("PROPFIND Response Body: {}", body);
289
290 self.parse_propfind_response(&body, path)
291 }
292
293 async fn upload(
295 &self,
296 local_path: &Path,
297 remote_path: &str,
298 ) -> Result<UploadResult, SyncError> {
299 let url = self.get_full_url(remote_path);
300 let start_time = SystemTime::now();
301
302 let file_data = tokio::fs::read(local_path)
304 .await
305 .map_err(|e| SyncError::Io(e))?;
306
307 let file_size = file_data.len() as u64;
308
309 let response = self
311 .client
312 .put(&url)
313 .header("Authorization", self.create_auth_header())
314 .body(file_data)
315 .send()
316 .await
317 .map_err(|e| SyncError::Network(e))?;
318
319 if !response.status().is_success() {
320 return Err(SyncError::Provider(ProviderError::ApiError(format!(
321 "Upload failed: {}",
322 response.status()
323 ))));
324 }
325
326 let elapsed = SystemTime::now()
327 .duration_since(start_time)
328 .unwrap_or(Duration::from_secs(0));
329
330 Ok(UploadResult {
331 bytes_uploaded: file_size,
332 file_size,
333 checksum: None,
334 elapsed_time: elapsed,
335 })
336 }
337
338 #[instrument(skip(self), fields(remote_path = %remote_path, local_path = %local_path.display()))]
340 async fn download(
341 &self,
342 remote_path: &str,
343 local_path: &Path,
344 ) -> Result<DownloadResult, SyncError> {
345 info!("开始下载文件");
346 let url = self.get_full_url(remote_path);
347 let start_time = SystemTime::now();
348
349 debug!("发送 GET 请求");
350 let response = self
351 .client
352 .get(&url)
353 .header("Authorization", self.create_auth_header())
354 .send()
355 .await
356 .map_err(|e| {
357 error!(error = %e, "下载请求失败");
358 SyncError::Network(e)
359 })?;
360
361 let status = response.status();
362 debug!(status = %status, "收到下载响应");
363
364 if !status.is_success() {
365 warn!(status = %status, "文件不存在或下载失败");
366 return Err(SyncError::Provider(ProviderError::FileNotFound(
367 remote_path.to_string(),
368 )));
369 }
370
371 let bytes = response.bytes().await.map_err(|e| {
372 error!(error = %e, "读取响应数据失败");
373 SyncError::Network(e)
374 })?;
375
376 let file_size = bytes.len() as u64;
377 debug!(file_size = %file_size, "下载数据大小: {} 字节", file_size);
378
379 if let Some(parent) = local_path.parent() {
381 debug!(parent = %parent.display(), "创建父目录");
382 tokio::fs::create_dir_all(parent).await.map_err(|e| {
383 error!(error = %e, "创建父目录失败");
384 SyncError::Io(e)
385 })?;
386 }
387
388 debug!("写入本地文件");
390 tokio::fs::write(local_path, bytes).await.map_err(|e| {
391 error!(error = %e, "写入本地文件失败");
392 SyncError::Io(e)
393 })?;
394
395 let elapsed = SystemTime::now()
396 .duration_since(start_time)
397 .unwrap_or(Duration::from_secs(0));
398
399 let speed = if elapsed.as_secs() > 0 {
400 file_size as f64 / elapsed.as_secs_f64() / 1024.0 / 1024.0
401 } else {
402 0.0
403 };
404
405 info!(
406 file_size = %file_size,
407 elapsed_ms = elapsed.as_millis(),
408 speed_mbps = %format!("{:.2}", speed),
409 "文件下载成功: {} 字节,耗时 {} ms,速度 {:.2} MB/s",
410 file_size, elapsed.as_millis(), speed
411 );
412
413 Ok(DownloadResult {
414 bytes_downloaded: file_size,
415 file_size,
416 checksum: None,
417 elapsed_time: elapsed,
418 })
419 }
420
421 #[instrument(skip(self), fields(path = %path))]
423 async fn delete(&self, path: &str) -> Result<(), SyncError> {
424 info!("开始删除文件或目录");
425 let url = self.get_full_url(path);
426
427 debug!("发送 DELETE 请求");
428 let response = self
429 .client
430 .delete(&url)
431 .header("Authorization", self.create_auth_header())
432 .send()
433 .await
434 .map_err(|e| {
435 error!(error = %e, "删除请求失败");
436 SyncError::Network(e)
437 })?;
438
439 let status = response.status();
440 debug!(status = %status, "收到删除响应");
441
442 if status.is_success() {
443 info!("删除成功");
444 Ok(())
445 } else if status == StatusCode::NOT_FOUND {
446 warn!("文件或目录不存在,视为删除成功");
447 Ok(())
448 } else {
449 error!(status = %status, "删除失败");
450 Err(SyncError::Provider(ProviderError::ApiError(format!(
451 "Delete failed: {}",
452 status
453 ))))
454 }
455 }
456
457 #[instrument(skip(self), fields(path = %path))]
459 async fn mkdir(&self, path: &str) -> Result<(), SyncError> {
460 info!("开始创建目录");
461 let url = self.get_full_url(path);
462
463 debug!("发送 MKCOL 请求");
464 let response = self
465 .client
466 .request(Method::from_bytes(b"MKCOL").unwrap(), &url)
467 .header("Authorization", self.create_auth_header())
468 .send()
469 .await
470 .map_err(|e| {
471 error!(error = %e, "创建目录请求失败");
472 SyncError::Network(e)
473 })?;
474
475 let status = response.status();
476 debug!(status = %status, "收到 MKCOL 响应");
477
478 if status.is_success() {
479 info!("目录创建成功");
480 Ok(())
481 } else if status == StatusCode::METHOD_NOT_ALLOWED {
482 warn!("目录可能已存在,视为创建成功");
484 Ok(())
485 } else {
486 error!(status = %status, "创建目录失败");
487 Err(SyncError::Provider(ProviderError::ApiError(format!(
488 "MKCOL failed: {}",
489 status
490 ))))
491 }
492 }
493
494 #[instrument(skip(self), fields(path = %path))]
496 async fn stat(&self, path: &str) -> Result<FileInfo, SyncError> {
497 debug!("查询文件或目录信息");
498 let url = self.get_full_url(path);
499
500 let response = self
501 .client
502 .request(Method::from_bytes(b"PROPFIND").unwrap(), &url)
503 .header("Authorization", self.create_auth_header())
504 .header("Depth", "0")
505 .send()
506 .await
507 .map_err(|e| {
508 error!(error = %e, "PROPFIND 请求失败");
509 SyncError::Network(e)
510 })?;
511
512 let status = response.status();
513 debug!(status = %status, "收到 stat 响应");
514
515 if !status.is_success() {
516 warn!("文件或目录不存在");
517 return Err(SyncError::Provider(ProviderError::FileNotFound(
518 path.to_string(),
519 )));
520 }
521
522 let is_dir = path.ends_with('/');
523 debug!(is_dir = %is_dir, "查询成功");
524
525 Ok(FileInfo {
526 path: path.to_string(),
527 size: 0,
528 modified: SystemTime::now()
529 .duration_since(UNIX_EPOCH)
530 .unwrap()
531 .as_secs() as i64,
532 hash: None,
533 is_dir,
534 })
535 }
536
537 #[instrument(skip(self), fields(path = %path))]
539 async fn exists(&self, path: &str) -> Result<bool, SyncError> {
540 debug!("检查文件或目录是否存在");
541 match self.stat(path).await {
542 Ok(_) => {
543 debug!("文件或目录存在");
544 Ok(true)
545 }
546 Err(SyncError::Provider(ProviderError::FileNotFound(_))) => {
547 debug!("文件或目录不存在");
548 Ok(false)
549 }
550 Err(e) => {
551 warn!(error = %e, "检查存在性时发生错误");
552 Err(e)
553 }
554 }
555 }
556}
557
558#[cfg(test)]
559mod tests {
560 use super::*;
561 use std::collections::HashMap;
562
563 #[test]
564 fn test_get_full_url() {
565 let config = AccountConfig {
566 id: "test".to_string(),
567 provider: crate::config::ProviderType::WebDAV,
568 name: "test".to_string(),
569 credentials: {
570 let mut creds = HashMap::new();
571 creds.insert("url".to_string(), "http://localhost:8080/dav".to_string());
572 creds.insert("username".to_string(), "user".to_string());
573 creds.insert("password".to_string(), "pass".to_string());
574 creds
575 },
576 rate_limit: None,
577 retry_policy: crate::config::RetryPolicy::default(),
578 };
579
580 let runtime = tokio::runtime::Runtime::new().unwrap();
581 let provider = runtime.block_on(WebDavProvider::new(&config)).unwrap();
582
583 assert_eq!(
584 provider.get_full_url("/test/file.txt"),
585 "http://localhost:8080/dav/test/file.txt"
586 );
587 assert_eq!(
588 provider.get_full_url("test/file.txt"),
589 "http://localhost:8080/dav/test/file.txt"
590 );
591
592 assert_eq!(
594 provider.get_full_url("/test/file with spaces.txt"),
595 "http://localhost:8080/dav/test/file%20with%20spaces.txt"
596 );
597 assert_eq!(
598 provider.get_full_url("/test/special{}.txt"),
599 "http://localhost:8080/dav/test/special%7B%7D.txt"
600 );
601 }
602
603 #[test]
604 fn test_auth_header() {
605 let config = AccountConfig {
606 id: "test".to_string(),
607 provider: crate::config::ProviderType::WebDAV,
608 name: "test".to_string(),
609 credentials: {
610 let mut creds = HashMap::new();
611 creds.insert("url".to_string(), "http://localhost:8080".to_string());
612 creds.insert("username".to_string(), "testuser".to_string());
613 creds.insert("password".to_string(), "testpass".to_string());
614 creds
615 },
616 rate_limit: None,
617 retry_policy: crate::config::RetryPolicy::default(),
618 };
619
620 let runtime = tokio::runtime::Runtime::new().unwrap();
621 let provider = runtime.block_on(WebDavProvider::new(&config)).unwrap();
622
623 let auth = provider.create_auth_header();
624 assert!(auth.starts_with("Basic "));
625
626 let encoded = auth.strip_prefix("Basic ").unwrap();
628 let decoded = base64::engine::general_purpose::STANDARD
629 .decode(encoded)
630 .unwrap();
631 assert_eq!(String::from_utf8(decoded).unwrap(), "testuser:testpass");
632 }
633
634 #[cfg(test)]
636 mod integration {
637 use super::*;
638 use std::env;
639 use std::net::SocketAddr;
640 use std::sync::Arc;
641 use tokio::sync::RwLock;
642
643 #[derive(Debug, Clone)]
644 struct InMemoryFile {
645 content: Vec<u8>,
646 is_dir: bool,
647 }
648
649 type FileStore = Arc<RwLock<HashMap<String, InMemoryFile>>>;
650
651 async fn start_mock_server() -> (SocketAddr, FileStore) {
652 use warp::Filter;
653
654 let store: FileStore = Arc::new(RwLock::new(HashMap::new()));
655
656 {
658 let mut files = store.write().await;
659 files.insert(
660 "/".to_string(),
661 InMemoryFile {
662 content: vec![],
663 is_dir: true,
664 },
665 );
666 }
667
668 let store_clone = store.clone();
669
670 let put_route = warp::put()
672 .and(warp::path::full())
673 .and(warp::body::bytes())
674 .and_then({
675 let store = store_clone.clone();
676 move |path: warp::path::FullPath, body: bytes::Bytes| {
677 let store = store.clone();
678 async move {
679 let path_str = path.as_str().to_string();
680 let mut files = store.write().await;
681
682 files.insert(
683 path_str,
684 InMemoryFile {
685 content: body.to_vec(),
686 is_dir: false,
687 },
688 );
689
690 Ok::<_, warp::Rejection>(warp::reply::with_status(
691 String::new(),
692 warp::http::StatusCode::CREATED,
693 ))
694 }
695 }
696 });
697
698 let get_route = warp::get().and(warp::path::full()).and_then({
700 let store = store_clone.clone();
701 move |path: warp::path::FullPath| {
702 let store = store.clone();
703 async move {
704 let path_str = path.as_str();
705 let files = store.read().await;
706
707 if let Some(file) = files.get(path_str) {
708 if !file.is_dir {
709 return Ok::<_, warp::Rejection>(warp::reply::with_status(
710 file.content.clone(),
711 warp::http::StatusCode::OK,
712 ));
713 }
714 }
715
716 Ok(warp::reply::with_status(
717 vec![],
718 warp::http::StatusCode::NOT_FOUND,
719 ))
720 }
721 }
722 });
723
724 let routes = put_route.or(get_route);
725 let (addr, server) = warp::serve(routes).bind_ephemeral(([127, 0, 0, 1], 0));
726 tokio::spawn(server);
727
728 (addr, store)
729 }
730
731 #[tokio::test]
732 async fn test_upload_download() {
733 let (addr, _store) = start_mock_server().await;
734
735 let config = AccountConfig {
736 id: "test".to_string(),
737 provider: crate::config::ProviderType::WebDAV,
738 name: "test".to_string(),
739 credentials: {
740 let mut creds = HashMap::new();
741 creds.insert("url".to_string(), format!("http://{}", addr));
742 creds.insert("username".to_string(), "test".to_string());
743 creds.insert("password".to_string(), "test".to_string());
744 creds
745 },
746 rate_limit: None,
747 retry_policy: crate::config::RetryPolicy::default(),
748 };
749
750 let provider = WebDavProvider::new(&config).await.unwrap();
751
752 let temp_dir = env::temp_dir();
754 let test_file = temp_dir.join("webdav_test_upload.txt");
755 let test_content = b"Hello WebDAV";
756 tokio::fs::write(&test_file, test_content).await.unwrap();
757
758 let upload_result = provider.upload(&test_file, "/test.txt").await.unwrap();
760 assert_eq!(upload_result.file_size, test_content.len() as u64);
761
762 let download_file = temp_dir.join("webdav_test_download.txt");
764 let download_result = provider
765 .download("/test.txt", &download_file)
766 .await
767 .unwrap();
768 assert_eq!(download_result.file_size, test_content.len() as u64);
769
770 let downloaded = tokio::fs::read(&download_file).await.unwrap();
772 assert_eq!(&downloaded, test_content);
773
774 tokio::fs::remove_file(&test_file).await.ok();
776 tokio::fs::remove_file(&download_file).await.ok();
777 }
778
779 #[tokio::test]
780 async fn test_large_file() {
781 let (addr, _store) = start_mock_server().await;
782
783 let config = AccountConfig {
784 id: "test".to_string(),
785 provider: crate::config::ProviderType::WebDAV,
786 name: "test".to_string(),
787 credentials: {
788 let mut creds = HashMap::new();
789 creds.insert("url".to_string(), format!("http://{}", addr));
790 creds.insert("username".to_string(), "test".to_string());
791 creds.insert("password".to_string(), "test".to_string());
792 creds
793 },
794 rate_limit: None,
795 retry_policy: crate::config::RetryPolicy::default(),
796 };
797
798 let provider = WebDavProvider::new(&config).await.unwrap();
799
800 let temp_dir = env::temp_dir();
802 let test_file = temp_dir.join("webdav_test_large.bin");
803 let large_content = vec![0u8; 1024 * 1024];
804 tokio::fs::write(&test_file, &large_content).await.unwrap();
805
806 let upload_result = provider.upload(&test_file, "/large.bin").await.unwrap();
808 assert_eq!(upload_result.file_size, large_content.len() as u64);
809
810 tokio::fs::remove_file(&test_file).await.ok();
812 }
813 }
814}