1#[cfg(not(target_family = "wasm"))]
15use std::{
16 io::Write,
17 path::{Path, PathBuf},
18};
19
20use bon::bon;
21#[cfg(not(target_family = "wasm"))]
22use futures::TryStreamExt;
23use futures::stream::{Stream, StreamExt};
24#[cfg(not(target_family = "wasm"))]
25use reqwest::header::IF_NONE_MATCH;
26#[cfg(not(target_family = "wasm"))]
27use serde::Deserialize;
28
29use super::files::extract_file_size;
30#[cfg(not(target_family = "wasm"))]
31use super::{
32 FileMetadataInfo,
33 files::{extract_commit_hash, extract_etag, extract_xet_hash, matches_any_glob},
34};
35use super::{HFRepository, RepoTreeEntry, RepoType};
36#[cfg(not(target_family = "wasm"))]
37use crate::cache::storage as cache;
38use crate::error::{HFError, HFResult};
39use crate::progress::{DownloadEvent, EmitEvent, FileProgress, FileStatus, Progress};
40use crate::{constants, retry};
41
42#[cfg(not(target_family = "wasm"))]
47pub type HFByteStream = Box<dyn Stream<Item = HFResult<bytes::Bytes>> + Send + Unpin>;
48#[cfg(target_family = "wasm")]
49pub type HFByteStream = Box<dyn Stream<Item = HFResult<bytes::Bytes>> + Unpin>;
50
51#[cfg(not(target_family = "wasm"))]
53struct DownloadFileParams {
54 filename: String,
55 local_dir: Option<PathBuf>,
56 revision: Option<String>,
57 force_download: bool,
58 local_files_only: bool,
59 progress: Option<Progress>,
60}
61
62struct DownloadFileStreamParams {
64 filename: String,
65 revision: Option<String>,
66 range: Option<std::ops::Range<u64>>,
67 progress: Option<Progress>,
68}
69
70#[cfg(not(target_family = "wasm"))]
72struct SnapshotDownloadParams {
73 revision: Option<String>,
74 allow_patterns: Option<Vec<String>>,
75 ignore_patterns: Option<Vec<String>>,
76 local_dir: Option<PathBuf>,
77 force_download: bool,
78 local_files_only: bool,
79 max_workers: Option<usize>,
80 progress: Option<Progress>,
81}
82
83impl<T: RepoType> HFRepository<T> {
84 #[cfg(not(target_family = "wasm"))]
85 async fn download_file_impl(&self, params: DownloadFileParams) -> HFResult<PathBuf> {
86 let result = self.download_file_inner(¶ms).await;
87 if result.is_ok() {
88 params.progress.emit(DownloadEvent::Complete);
89 }
90 result
91 }
92
93 #[cfg(not(target_family = "wasm"))]
94 async fn download_file_inner(&self, params: &DownloadFileParams) -> HFResult<PathBuf> {
95 if params.local_dir.is_some() {
96 self.download_file_to_local_dir(params).await
97 } else {
98 if !self.hf_client.cache_enabled() {
99 return Err(HFError::CacheNotEnabled);
100 }
101 self.download_file_to_cache(params).await
102 }
103 }
104
105 #[cfg(not(target_family = "wasm"))]
117 async fn resolve_xet_hash_and_size(
118 &self,
119 revision: &str,
120 filename: &str,
121 ) -> HFResult<(Option<String>, Option<u64>)> {
122 let repo_path = self.repo_path();
123 let url = self
124 .hf_client
125 .download_url(T::default().url_prefix(), &repo_path, revision, filename)?;
126 let headers = self.hf_client.auth_headers();
127 let head_response = retry::retry(self.hf_client.retry_config(), || {
128 self.hf_client.http_client().head(&url).headers(headers.clone()).send()
129 })
130 .await?;
131 let head_response = self
132 .hf_client
133 .check_response(
134 head_response,
135 Some(&repo_path),
136 crate::error::NotFoundContext::Entry {
137 path: filename.to_string(),
138 },
139 )
140 .await?;
141 Ok((extract_xet_hash(&head_response), extract_file_size(&head_response)))
142 }
143
144 #[cfg(target_family = "wasm")]
145 async fn resolve_xet_hash_and_size(
146 &self,
147 revision: &str,
148 filename: &str,
149 ) -> HFResult<(Option<String>, Option<u64>)> {
150 let entries = self
151 .get_paths_info()
152 .paths(vec![filename.to_string()])
153 .revision(revision.to_string())
154 .send()
155 .await?;
156 let entry = entries
157 .into_iter()
158 .find(|e| matches!(e, RepoTreeEntry::File { path, .. } if path == filename));
159 match entry {
160 Some(RepoTreeEntry::File { xet_hash, size, .. }) => Ok((xet_hash, Some(size))),
161 _ => Err(HFError::EntryNotFound {
162 path: filename.to_string(),
163 repo_id: self.repo_path(),
164 context: None,
165 }),
166 }
167 }
168
169 async fn download_file_stream_impl(
170 &self,
171 params: DownloadFileStreamParams,
172 ) -> HFResult<(Option<u64>, HFByteStream)> {
173 if let Some(ref range) = params.range
174 && range.start >= range.end
175 {
176 return Err(HFError::InvalidParameter(format!(
177 "range start ({}) must be less than end ({})",
178 range.start, range.end
179 )));
180 }
181
182 let revision = params.revision.as_deref().unwrap_or(constants::DEFAULT_REVISION);
183 let repo_path = self.repo_path();
184 let url = self
185 .hf_client
186 .download_url(self.repo_type.url_prefix(), &repo_path, revision, ¶ms.filename)?;
187
188 let headers = self.hf_client.auth_headers();
189
190 let (xet_hash, file_size_hint) = self.resolve_xet_hash_and_size(revision, ¶ms.filename).await?;
191
192 if let Some(xet_hash) = xet_hash {
193 let file_size: u64 = file_size_hint.unwrap_or_else(|| {
194 tracing::warn!(url = %url, "missing file size for xet file, defaulting to 0");
195 0
196 });
197
198 let content_length = params.range.as_ref().map(|r| r.end.saturating_sub(r.start)).or(Some(file_size));
199
200 let stream = self
201 .xet_download_stream(revision, &xet_hash, file_size, params.range.clone())
202 .await?;
203
204 let total_bytes = content_length.unwrap_or(0);
205 params.progress.emit(DownloadEvent::Start {
206 total_files: 1,
207 total_bytes,
208 });
209 let wrapped =
210 wrap_stream_with_progress(Box::new(Box::pin(stream)), params.progress, params.filename, total_bytes);
211 #[cfg(target_family = "wasm")]
212 let wrapped = buffer_wasm_stream(wrapped);
213 return Ok((content_length, wrapped));
214 }
215
216 let range_header = params
217 .range
218 .as_ref()
219 .map(|r| format!("bytes={}-{}", r.start, r.end.saturating_sub(1)));
220 let response = retry::retry(self.hf_client.retry_config(), || {
221 let mut req = self.hf_client.http_client().get(&url).headers(headers.clone());
222 if let Some(ref range) = range_header {
223 req = req.header(reqwest::header::RANGE, range);
224 }
225 req.send()
226 })
227 .await?;
228 let response = self
229 .hf_client
230 .check_response(
231 response,
232 Some(&repo_path),
233 crate::error::NotFoundContext::Entry {
234 path: params.filename.clone(),
235 },
236 )
237 .await?;
238
239 let content_length = extract_file_size(&response);
240 let total_bytes = content_length.unwrap_or(0);
241 let stream = response.bytes_stream().map(|r| r.map_err(HFError::from));
242 params.progress.emit(DownloadEvent::Start {
243 total_files: 1,
244 total_bytes,
245 });
246 let wrapped =
247 wrap_stream_with_progress(Box::new(Box::pin(stream)), params.progress, params.filename, total_bytes);
248 #[cfg(target_family = "wasm")]
249 let wrapped = buffer_wasm_stream(wrapped);
250 Ok((content_length, wrapped))
251 }
252
253 async fn download_file_to_bytes_impl(&self, params: DownloadFileStreamParams) -> HFResult<bytes::Bytes> {
254 let (content_length, stream) = self.download_file_stream_impl(params).await?;
255 futures::pin_mut!(stream);
256
257 let capacity = content_length.unwrap_or(0) as usize;
258 let mut buf = bytes::BytesMut::with_capacity(capacity);
259 while let Some(chunk) = stream.next().await {
260 buf.extend_from_slice(&chunk?);
261 }
262 Ok(buf.freeze())
263 }
264
265 #[cfg(not(target_family = "wasm"))]
266 async fn download_file_to_local_dir(&self, params: &DownloadFileParams) -> HFResult<PathBuf> {
267 let revision = params.revision.as_deref().unwrap_or(constants::DEFAULT_REVISION);
268 let repo_path = self.repo_path();
269 let url = self
270 .hf_client
271 .download_url(self.repo_type.url_prefix(), &repo_path, revision, ¶ms.filename)?;
272
273 let headers = self.hf_client.auth_headers();
274 let head_response = retry::retry(self.hf_client.retry_config(), || {
275 self.hf_client.http_client().head(&url).headers(headers.clone()).send()
276 })
277 .await?;
278
279 let head_response = self
280 .hf_client
281 .check_response(
282 head_response,
283 Some(&repo_path),
284 crate::error::NotFoundContext::Entry {
285 path: params.filename.clone(),
286 },
287 )
288 .await?;
289
290 let file_size = extract_file_size(&head_response).unwrap_or(0);
291 let has_xet_hash = head_response.headers().get(constants::HEADER_X_XET_HASH).is_some();
292
293 params.progress.emit(DownloadEvent::Start {
294 total_files: 1,
295 total_bytes: file_size,
296 });
297
298 if has_xet_hash {
299 let local_dir = params.local_dir.as_ref().unwrap();
300 return self
301 .xet_download_to_local_dir(revision, ¶ms.filename, local_dir, &head_response, ¶ms.progress)
302 .await;
303 }
304
305 let response = retry::retry(self.hf_client.retry_config(), || {
306 self.hf_client.http_client().get(&url).headers(headers.clone()).send()
307 })
308 .await?;
309 let response = self
310 .hf_client
311 .check_response(
312 response,
313 Some(&repo_path),
314 crate::error::NotFoundContext::Entry {
315 path: params.filename.clone(),
316 },
317 )
318 .await?;
319
320 let local_dir = params.local_dir.as_ref().unwrap();
321 std::fs::create_dir_all(local_dir)?;
322
323 let dest_path = local_dir.join(¶ms.filename);
324 if let Some(parent) = dest_path.parent() {
325 std::fs::create_dir_all(parent)?;
326 }
327
328 stream_response_to_file_with_progress(
329 response,
330 &dest_path,
331 ¶ms.progress,
332 Some(¶ms.filename),
333 file_size,
334 )
335 .await?;
336 params.progress.emit(DownloadEvent::Progress {
337 files: vec![FileProgress {
338 filename: params.filename.clone(),
339 bytes_completed: file_size,
340 total_bytes: file_size,
341 status: FileStatus::Complete,
342 }],
343 });
344
345 Ok(dest_path)
346 }
347
348 #[cfg(not(target_family = "wasm"))]
352 fn resolve_from_cache_only(&self, repo_folder: &str, revision: &str, filename: &str) -> HFResult<PathBuf> {
353 let cache_dir = self.hf_client.cache_dir();
354
355 let commit_hash = if cache::is_commit_hash(revision) {
356 Some(revision.to_string())
357 } else {
358 let ref_path = cache::ref_path(cache_dir, repo_folder, revision);
359 std::fs::read_to_string(&ref_path).ok().map(|s| s.trim().to_string())
360 };
361
362 if let Some(ref hash) = commit_hash {
363 let snap = cache::snapshot_path(cache_dir, repo_folder, hash, filename);
364 if snap.exists() {
365 return Ok(snap);
366 }
367 if cache::no_exist_path(cache_dir, repo_folder, hash, filename).exists() {
368 return Err(HFError::EntryNotFound {
369 path: filename.to_string(),
370 repo_id: String::new(),
371 context: None,
372 });
373 }
374 }
375
376 Err(HFError::LocalEntryNotFound {
377 path: filename.to_string(),
378 })
379 }
380
381 #[cfg(not(target_family = "wasm"))]
385 fn find_cached_etag(&self, repo_folder: &str, revision: &str, filename: &str) -> Option<String> {
386 let cache_dir = self.hf_client.cache_dir();
387
388 let commit_hash = if cache::is_commit_hash(revision) {
389 Some(revision.to_string())
390 } else {
391 let ref_path = cache::ref_path(cache_dir, repo_folder, revision);
392 std::fs::read_to_string(&ref_path).ok().map(|s| s.trim().to_string())
393 };
394
395 let hash = commit_hash?;
396 let snap = cache::snapshot_path(cache_dir, repo_folder, &hash, filename);
397 let target = std::fs::read_link(&snap).ok()?;
398 target.file_name()?.to_str().map(|s| s.to_string())
399 }
400
401 #[cfg(not(target_family = "wasm"))]
402 async fn download_file_to_cache(&self, params: &DownloadFileParams) -> HFResult<PathBuf> {
403 let revision = params.revision.as_deref().unwrap_or(constants::DEFAULT_REVISION);
404 let cache_dir = self.hf_client.cache_dir();
405 let repo_folder = cache::repo_folder_name(&self.repo_path(), self.repo_type.plural());
406 let force_download = params.force_download;
407
408 if cache::is_commit_hash(revision) && !force_download {
409 let snap = cache::snapshot_path(cache_dir, &repo_folder, revision, ¶ms.filename);
410 if snap.exists() {
411 return Ok(snap);
412 }
413 }
414
415 if params.local_files_only {
416 return self.resolve_from_cache_only(&repo_folder, revision, ¶ms.filename);
417 }
418
419 let result = self
420 .download_file_to_cache_network(params, revision, cache_dir, &repo_folder, force_download)
421 .await;
422
423 match &result {
424 Err(e) if e.is_transient() && !force_download => self
425 .resolve_from_cache_only(&repo_folder, revision, ¶ms.filename)
426 .or(result),
427 _ => result,
428 }
429 }
430
431 #[cfg(not(target_family = "wasm"))]
432 async fn download_file_to_cache_network(
433 &self,
434 params: &DownloadFileParams,
435 revision: &str,
436 cache_dir: &Path,
437 repo_folder: &str,
438 force_download: bool,
439 ) -> HFResult<PathBuf> {
440 let repo_path = self.repo_path();
441 let url = self
442 .hf_client
443 .download_url(self.repo_type.url_prefix(), &repo_path, revision, ¶ms.filename)?;
444
445 let cached_etag = if !force_download {
446 self.find_cached_etag(repo_folder, revision, ¶ms.filename)
447 } else {
448 None
449 };
450
451 let mut head_headers = self.hf_client.auth_headers();
452 if let Some(ref etag_val) = cached_etag
453 && let Ok(hv) = reqwest::header::HeaderValue::from_str(&format!("\"{etag_val}\""))
454 {
455 head_headers.insert(IF_NONE_MATCH, hv);
456 }
457
458 let head_response = self.hf_client.head_with_relative_redirects(&url, &head_headers).await?;
459
460 let status = head_response.status();
461
462 if status == reqwest::StatusCode::NOT_FOUND {
463 return Err(mark_no_exist_and_return_error(
464 cache_dir,
465 repo_folder,
466 revision,
467 &head_response,
468 &repo_path,
469 ¶ms.filename,
470 )
471 .await);
472 }
473
474 if status == reqwest::StatusCode::NOT_MODIFIED {
475 let etag = cached_etag
476 .ok_or_else(|| HFError::malformed_response_at("304 Not Modified without cached ETag", url.clone()))?;
477 let commit_hash = if cache::is_commit_hash(revision) {
478 revision.to_string()
479 } else {
480 cache::read_ref(cache_dir, repo_folder, revision).await?.ok_or_else(|| {
481 HFError::malformed_response_at("304 Not Modified without cached commit hash", url.clone())
482 })?
483 };
484 return finalize_cached_file(cache_dir, repo_folder, revision, &commit_hash, ¶ms.filename, &etag).await;
485 }
486
487 let etag = extract_etag(&head_response)
488 .ok_or_else(|| HFError::malformed_response_at("missing ETag header", url.clone()));
489 let commit_hash = extract_commit_hash(&head_response);
490 let xet_hash = extract_xet_hash(&head_response);
491 let has_xet_hash = xet_hash.is_some();
492 let file_size: u64 = extract_file_size(&head_response).unwrap_or_else(|| {
493 tracing::warn!(url = %url, "missing or invalid Content-Length/X-Linked-Size header, defaulting file size to 0");
494 0
495 });
496
497 if !status.is_success() && !status.is_redirection() {
498 self.hf_client
499 .check_response(
500 head_response,
501 Some(&repo_path),
502 crate::error::NotFoundContext::Entry {
503 path: params.filename.clone(),
504 },
505 )
506 .await?;
507 }
508
509 let etag = etag?;
510 let commit_hash =
511 commit_hash.ok_or_else(|| HFError::malformed_response_at("missing X-Repo-Commit header", url.clone()))?;
512
513 params.progress.emit(DownloadEvent::Start {
514 total_files: 1,
515 total_bytes: file_size,
516 });
517
518 if has_xet_hash {
519 let xet_hash =
520 xet_hash.ok_or_else(|| HFError::malformed_response_at("missing X-Xet-Hash header", url.clone()))?;
521 let blob = cache::blob_path(cache_dir, repo_folder, &etag);
522 if !blob.exists() || force_download {
523 if let Some(parent) = blob.parent() {
524 std::fs::create_dir_all(parent)?;
525 }
526 let _lock = cache::acquire_lock(cache_dir, repo_folder, &etag).await?;
527
528 self.xet_download_to_blob(revision, ¶ms.filename, &xet_hash, file_size, &blob, ¶ms.progress)
529 .await?;
530 }
531
532 return finalize_cached_file(cache_dir, repo_folder, revision, &commit_hash, ¶ms.filename, &etag).await;
533 }
534
535 let blob = cache::blob_path(cache_dir, repo_folder, &etag);
536
537 if blob.exists() && !force_download {
538 params.progress.emit(DownloadEvent::Progress {
539 files: vec![FileProgress {
540 filename: params.filename.clone(),
541 bytes_completed: file_size,
542 total_bytes: file_size,
543 status: FileStatus::Complete,
544 }],
545 });
546 return finalize_cached_file(cache_dir, repo_folder, revision, &commit_hash, ¶ms.filename, &etag).await;
547 }
548
549 let _lock = cache::acquire_lock(cache_dir, repo_folder, &etag).await?;
550 let incomplete_path = PathBuf::from(format!("{}.incomplete", blob.display()));
551 if let Some(parent) = incomplete_path.parent() {
552 std::fs::create_dir_all(parent)?;
553 }
554
555 let dl_headers = self.hf_client.auth_headers();
556 let response = retry::retry(self.hf_client.retry_config(), || {
557 self.hf_client.http_client().get(&url).headers(dl_headers.clone()).send()
558 })
559 .await?;
560 stream_response_to_file_with_progress(
561 response,
562 &incomplete_path,
563 ¶ms.progress,
564 Some(¶ms.filename),
565 file_size,
566 )
567 .await?;
568 params.progress.emit(DownloadEvent::Progress {
569 files: vec![FileProgress {
570 filename: params.filename.clone(),
571 bytes_completed: file_size,
572 total_bytes: file_size,
573 status: FileStatus::Complete,
574 }],
575 });
576 std::fs::rename(&incomplete_path, &blob)?;
577
578 finalize_cached_file(cache_dir, repo_folder, revision, &commit_hash, ¶ms.filename, &etag).await
579 }
580
581 #[cfg(not(target_family = "wasm"))]
582 async fn resolve_commit_hash(&self, revision: &str) -> HFResult<String> {
583 if cache::is_commit_hash(revision) {
584 return Ok(revision.to_string());
585 }
586 #[derive(Deserialize)]
587 struct ShaOnly {
588 sha: Option<String>,
589 }
590 let repo_path = self.repo_path();
591 let info: ShaOnly = self.fetch_repo_info(Some(revision.to_string()), None).await?;
592 info.sha.ok_or_else(|| {
593 HFError::malformed_response(format!("repo info for {}@{} returned no commit sha", repo_path, revision))
594 })
595 }
596
597 #[cfg(not(target_family = "wasm"))]
598 async fn list_filtered_files(
599 &self,
600 revision: &str,
601 allow_patterns: Option<&Vec<String>>,
602 ignore_patterns: Option<&Vec<String>>,
603 ) -> HFResult<Vec<String>> {
604 let stream = self.list_tree().revision(revision.to_string()).recursive(true).send()?;
605 futures::pin_mut!(stream);
606
607 let mut filenames: Vec<String> = Vec::new();
608 while let Some(entry) = stream.next().await {
609 let entry = entry?;
610 if let RepoTreeEntry::File { path, .. } = entry {
611 filenames.push(path);
612 }
613 }
614
615 if let Some(allow) = allow_patterns {
616 filenames.retain(|f| matches_any_glob(allow, f));
617 }
618 if let Some(ignore) = ignore_patterns {
619 filenames.retain(|f| !matches_any_glob(ignore, f));
620 }
621
622 Ok(filenames)
623 }
624
625 #[cfg(not(target_family = "wasm"))]
626 async fn snapshot_download_impl(&self, params: SnapshotDownloadParams) -> HFResult<PathBuf> {
627 if params.local_dir.is_none() && !self.hf_client.cache_enabled() {
628 return Err(HFError::CacheNotEnabled);
629 }
630 let revision = params.revision.as_deref().unwrap_or(constants::DEFAULT_REVISION);
631 let max_workers = params.max_workers.unwrap_or(8);
632 let repo_folder = cache::repo_folder_name(&self.repo_path(), self.repo_type.plural());
633 let cache_dir = self.hf_client.cache_dir();
634
635 if params.local_files_only {
636 let commit_hash = if cache::is_commit_hash(revision) {
637 revision.to_string()
638 } else {
639 cache::read_ref(cache_dir, &repo_folder, revision).await?.ok_or_else(|| {
640 HFError::LocalEntryNotFound {
641 path: format!("{}/{}", repo_folder, revision),
642 }
643 })?
644 };
645 let snapshot_dir = cache_dir.join(&repo_folder).join("snapshots").join(&commit_hash);
646 if snapshot_dir.exists() {
647 return Ok(snapshot_dir);
648 }
649 return Err(HFError::LocalEntryNotFound {
650 path: format!("{}/{}", repo_folder, commit_hash),
651 });
652 }
653
654 let commit_hash = self.resolve_commit_hash(revision).await?;
655
656 let mut filenames = self
657 .list_filtered_files(&commit_hash, params.allow_patterns.as_ref(), params.ignore_patterns.as_ref())
658 .await?;
659
660 let total_files = filenames.len();
661 let force = params.force_download;
662
663 let mut cached_filenames = Vec::new();
664 if !force && params.local_dir.is_none() {
665 filenames.retain(|f| {
666 if cache::snapshot_path(cache_dir, &repo_folder, &commit_hash, f).exists() {
667 cached_filenames.push(f.clone());
668 false
669 } else {
670 true
671 }
672 });
673 }
674
675 let repo_path = self.repo_path();
676 let repo_path_ref = &repo_path;
677 let commit_hash_ref = &commit_hash;
678 let mut head_futs = Vec::with_capacity(filenames.len());
679 for filename in &filenames {
680 let auth = self.hf_client.auth_headers();
681 let filename = filename.clone();
682 let repo_folder_ref = &repo_folder;
683 head_futs.push(async move {
684 let url = self
685 .hf_client
686 .download_url(self.repo_type.url_prefix(), repo_path_ref, commit_hash_ref, &filename)?;
687 let resp = self.hf_client.head_with_relative_redirects(&url, &auth).await?;
688 if resp.status() == reqwest::StatusCode::NOT_FOUND {
695 if let Some(commit) = extract_commit_hash(&resp) {
696 let no_exist = cache::no_exist_path(cache_dir, repo_folder_ref, &commit, &filename);
697 if let Some(parent) = no_exist.parent() {
698 let _ = std::fs::create_dir_all(parent);
699 }
700 let _ = std::fs::write(&no_exist, b"");
701 }
702 return Ok::<_, HFError>(None);
703 } else if !resp.status().is_success() && !resp.status().is_redirection() {
704 let context = Box::new(crate::error::HttpErrorContext::from_response(resp).await);
705 return Err(HFError::Http { context });
706 }
707 let etag = extract_etag(&resp).ok_or_else(|| {
708 HFError::malformed_response_at(format!("missing ETag header for {filename}"), url.clone())
709 })?;
710 let commit = extract_commit_hash(&resp).unwrap_or_else(|| commit_hash_ref.clone());
711 let xet_hash = extract_xet_hash(&resp);
712 let file_size: u64 = extract_file_size(&resp).unwrap_or_else(|| {
713 tracing::warn!(file = %filename, "missing or invalid Content-Length/X-Linked-Size header, defaulting file size to 0");
714 0
715 });
716 let location = Some(resp.url().to_string());
717 Ok::<_, HFError>(Some(FileMetadataInfo {
718 filename,
719 etag,
720 commit_hash: commit,
721 xet_hash,
722 file_size,
723 location,
724 }))
725 });
726 }
727
728 let file_metas: Vec<FileMetadataInfo> = futures::stream::iter(head_futs)
729 .buffer_unordered(max_workers)
730 .try_collect::<Vec<Option<FileMetadataInfo>>>()
731 .await?
732 .into_iter()
733 .flatten()
734 .collect();
735
736 let total_bytes: u64 = file_metas.iter().map(|m| m.file_size).sum();
737 params.progress.emit(DownloadEvent::Start {
738 total_files,
739 total_bytes,
740 });
741 if !cached_filenames.is_empty() {
742 params.progress.emit(DownloadEvent::Progress {
743 files: cached_filenames
744 .iter()
745 .map(|f| FileProgress {
746 filename: f.clone(),
747 bytes_completed: 0,
748 total_bytes: 0,
749 status: FileStatus::Complete,
750 })
751 .collect(),
752 });
753 }
754
755 let mut xet_metas = Vec::new();
756 let mut non_xet_filenames = Vec::new();
757
758 if let Some(ref local_dir) = params.local_dir {
759 let mut local_cached = Vec::new();
760 for meta in file_metas {
761 let dest = local_dir.join(&meta.filename);
762 if dest.exists() && !force {
763 local_cached.push(meta.filename);
764 continue;
765 }
766 if meta.xet_hash.is_some() {
767 xet_metas.push(meta);
768 } else {
769 non_xet_filenames.push(meta.filename);
770 }
771 }
772 if !local_cached.is_empty() {
773 params.progress.emit(DownloadEvent::Progress {
774 files: local_cached
775 .iter()
776 .map(|f| FileProgress {
777 filename: f.clone(),
778 bytes_completed: 0,
779 total_bytes: 0,
780 status: FileStatus::Complete,
781 })
782 .collect(),
783 });
784 }
785
786 let xet_batch_fut = async {
787 if xet_metas.is_empty() {
788 return Ok::<_, HFError>(());
789 }
790 let batch_files: Vec<crate::xet::XetBatchFile> = xet_metas
791 .iter()
792 .map(|m| crate::xet::XetBatchFile {
793 hash: m.xet_hash.as_ref().unwrap().clone(),
794 file_size: m.file_size,
795 path: local_dir.join(&m.filename),
796 filename: m.filename.clone(),
797 })
798 .collect();
799 self.xet_download_batch(&commit_hash, &batch_files, ¶ms.progress).await?;
800 Ok(())
801 };
802
803 let non_xet_dl_params = build_download_params(
804 &repo_path,
805 &non_xet_filenames,
806 &commit_hash,
807 params.force_download,
808 Some(local_dir.clone()),
809 ¶ms.progress,
810 );
811 let non_xet_fut = async {
812 download_concurrently(self, &non_xet_dl_params, max_workers).await?;
813 Ok::<_, HFError>(())
814 };
815
816 tokio::try_join!(xet_batch_fut, non_xet_fut)?;
817 params.progress.emit(DownloadEvent::Complete);
818 return Ok(local_dir.clone());
819 }
820
821 let mut cached_progress: Vec<FileProgress> = Vec::new();
823 for meta in file_metas {
824 let blob = cache::blob_path(cache_dir, &repo_folder, &meta.etag);
825 if blob.exists() && !force {
826 cache::create_pointer_symlink(cache_dir, &repo_folder, &meta.commit_hash, &meta.filename, &meta.etag)
827 .await?;
828 cached_progress.push(FileProgress {
829 filename: meta.filename.clone(),
830 bytes_completed: meta.file_size,
831 total_bytes: meta.file_size,
832 status: FileStatus::Complete,
833 });
834 continue;
835 }
836 if meta.xet_hash.is_some() {
837 xet_metas.push(meta);
838 } else {
839 non_xet_filenames.push(meta.filename);
840 }
841 }
842 if !cached_progress.is_empty() {
843 params.progress.emit(DownloadEvent::Progress { files: cached_progress });
844 }
845
846 let xet_batch_fut = async {
847 if xet_metas.is_empty() {
848 return Ok::<_, HFError>(());
849 }
850 let mut locks = Vec::with_capacity(xet_metas.len());
851 for m in &xet_metas {
852 locks.push(cache::acquire_lock(cache_dir, &repo_folder, &m.etag).await?);
853 }
854 let batch_files: Vec<crate::xet::XetBatchFile> = xet_metas
855 .iter()
856 .map(|m| crate::xet::XetBatchFile {
857 hash: m.xet_hash.as_ref().unwrap().clone(),
858 file_size: m.file_size,
859 path: cache::blob_path(cache_dir, &repo_folder, &m.etag),
860 filename: m.filename.clone(),
861 })
862 .collect();
863 self.xet_download_batch(&commit_hash, &batch_files, ¶ms.progress).await?;
864 for m in &xet_metas {
865 cache::create_pointer_symlink(cache_dir, &repo_folder, &m.commit_hash, &m.filename, &m.etag).await?;
866 }
867 drop(locks);
868 Ok(())
869 };
870
871 let non_xet_dl_params = build_download_params(
872 &repo_path,
873 &non_xet_filenames,
874 &commit_hash,
875 params.force_download,
876 None,
877 ¶ms.progress,
878 );
879 let non_xet_fut = async {
880 download_concurrently(self, &non_xet_dl_params, max_workers).await?;
881 Ok::<_, HFError>(())
882 };
883
884 tokio::try_join!(xet_batch_fut, non_xet_fut)?;
885
886 if !cache::is_commit_hash(revision) {
887 cache::write_ref(cache_dir, &repo_folder, revision, &commit_hash).await?;
888 }
889
890 params.progress.emit(DownloadEvent::Complete);
891 Ok(cache_dir.join(&repo_folder).join("snapshots").join(&commit_hash))
892 }
893}
894
895#[cfg(not(target_family = "wasm"))]
896async fn mark_no_exist_and_return_error(
897 cache_dir: &Path,
898 repo_folder: &str,
899 revision: &str,
900 response: &reqwest::Response,
901 repo_id: &str,
902 filename: &str,
903) -> HFError {
904 if let Some(commit_hash) = extract_commit_hash(response) {
905 let no_exist = cache::no_exist_path(cache_dir, repo_folder, &commit_hash, filename);
906 if let Some(parent) = no_exist.parent() {
907 let _ = std::fs::create_dir_all(parent);
908 }
909 let _ = std::fs::write(&no_exist, b"");
910 if !cache::is_commit_hash(revision) {
911 let _ = cache::write_ref(cache_dir, repo_folder, revision, &commit_hash).await;
912 }
913 }
914 HFError::EntryNotFound {
915 path: filename.to_string(),
916 repo_id: repo_id.to_string(),
917 context: None,
918 }
919}
920
921#[cfg(not(target_family = "wasm"))]
922async fn finalize_cached_file(
923 cache_dir: &Path,
924 repo_folder: &str,
925 revision: &str,
926 commit_hash: &str,
927 filename: &str,
928 etag: &str,
929) -> HFResult<PathBuf> {
930 if !cache::is_commit_hash(revision) {
931 cache::write_ref(cache_dir, repo_folder, revision, commit_hash).await?;
932 }
933 cache::create_pointer_symlink(cache_dir, repo_folder, commit_hash, filename, etag).await?;
934 Ok(cache::snapshot_path(cache_dir, repo_folder, commit_hash, filename))
935}
936
937#[cfg(not(target_family = "wasm"))]
938fn build_download_params(
939 _repo_id: &str,
940 filenames: &[String],
941 commit_hash: &str,
942 force_download: bool,
943 local_dir: Option<PathBuf>,
944 progress: &Option<Progress>,
945) -> Vec<DownloadFileParams> {
946 filenames
947 .iter()
948 .map(|filename| DownloadFileParams {
949 filename: filename.clone(),
950 local_dir: local_dir.clone(),
951 revision: Some(commit_hash.to_string()),
952 force_download,
953 local_files_only: false,
954 progress: progress.clone(),
955 })
956 .collect()
957}
958
959#[cfg(not(target_family = "wasm"))]
960async fn download_concurrently<T: RepoType>(
961 api: &HFRepository<T>,
962 params: &[DownloadFileParams],
963 max_workers: usize,
964) -> HFResult<Vec<PathBuf>> {
965 let mut download_futs = Vec::with_capacity(params.len());
966 for file_params in params {
967 download_futs.push(api.download_file_inner(file_params));
968 }
969 futures::stream::iter(download_futs)
970 .buffer_unordered(max_workers)
971 .try_collect()
972 .await
973}
974
975#[cfg(not(target_family = "wasm"))]
976async fn stream_response_to_file_with_progress(
977 response: reqwest::Response,
978 dest: &Path,
979 handler: &Option<Progress>,
980 filename: Option<&str>,
981 total_bytes: u64,
982) -> HFResult<()> {
983 let mut file = std::fs::File::create(dest)?;
984 let mut stream = response.bytes_stream();
985 let mut bytes_read: u64 = 0;
986
987 if let (Some(h), Some(filename)) = (handler, filename) {
988 h.emit(DownloadEvent::Progress {
989 files: vec![FileProgress {
990 filename: filename.to_string(),
991 bytes_completed: 0,
992 total_bytes,
993 status: FileStatus::Started,
994 }],
995 });
996 }
997
998 while let Some(chunk) = stream.next().await {
999 let chunk = chunk?;
1000 file.write_all(&chunk)?;
1001 bytes_read += chunk.len() as u64;
1002
1003 if let (Some(h), Some(filename)) = (handler, filename) {
1004 h.emit(DownloadEvent::Progress {
1005 files: vec![FileProgress {
1006 filename: filename.to_string(),
1007 bytes_completed: bytes_read,
1008 total_bytes,
1009 status: FileStatus::InProgress,
1010 }],
1011 });
1012 }
1013 }
1014 file.flush()?;
1015 Ok(())
1016}
1017
1018#[cfg(target_family = "wasm")]
1031pub(crate) fn buffer_wasm_stream(inner: HFByteStream) -> HFByteStream {
1032 use futures::SinkExt;
1033 use futures::channel::mpsc;
1034
1035 const BUFFER_DEPTH: usize = 2;
1036 let (mut tx, rx) = mpsc::channel::<HFResult<bytes::Bytes>>(BUFFER_DEPTH);
1037
1038 wasm_bindgen_futures::spawn_local(async move {
1039 let mut inner = inner;
1040 while let Some(item) = inner.next().await {
1041 let is_err = item.is_err();
1042 if tx.send(item).await.is_err() {
1043 return;
1044 }
1045 if is_err {
1046 return;
1047 }
1048 }
1049 });
1050
1051 Box::new(Box::pin(rx))
1052}
1053
1054pub(crate) fn wrap_stream_with_progress(
1055 stream: HFByteStream,
1056 progress: Option<Progress>,
1057 filename: String,
1058 total_bytes: u64,
1059) -> HFByteStream {
1060 if progress.is_none() {
1061 return stream;
1062 }
1063 let wrapped = futures::stream::unfold((stream, 0u64, false), move |(mut inner, bytes_completed, ended)| {
1064 let progress = progress.clone();
1065 let filename = filename.clone();
1066 async move {
1067 if ended {
1068 return None;
1069 }
1070 match inner.next().await {
1071 Some(Ok(chunk)) => {
1072 let bytes_completed = bytes_completed + chunk.len() as u64;
1073 progress.emit(DownloadEvent::Progress {
1074 files: vec![FileProgress {
1075 filename,
1076 bytes_completed,
1077 total_bytes,
1078 status: FileStatus::InProgress,
1079 }],
1080 });
1081 Some((Ok(chunk), (inner, bytes_completed, false)))
1082 },
1083 Some(Err(e)) => Some((Err(e), (inner, bytes_completed, true))),
1084 None => {
1085 progress.emit(DownloadEvent::Complete);
1086 None
1087 },
1088 }
1089 }
1090 });
1091 Box::new(Box::pin(wrapped))
1092}
1093
1094#[bon]
1095impl<T: RepoType> HFRepository<T> {
1096 #[cfg(not(target_family = "wasm"))]
1141 #[builder(finish_fn = send, derive(Debug, Clone))]
1142 pub async fn download_file(
1143 &self,
1144 #[builder(into)]
1146 filename: String,
1147 #[builder(into)]
1149 local_dir: Option<PathBuf>,
1150 #[builder(into)]
1152 revision: Option<String>,
1153 #[builder(default)]
1155 force_download: bool,
1156 #[builder(default)]
1158 local_files_only: bool,
1159 #[builder(into)]
1161 progress: Option<Progress>,
1162 ) -> HFResult<PathBuf> {
1163 Box::pin(self.download_file_impl(DownloadFileParams {
1164 filename,
1165 local_dir,
1166 revision,
1167 force_download,
1168 local_files_only,
1169 progress,
1170 }))
1171 .await
1172 }
1173
1174 #[builder(finish_fn = send, derive(Debug, Clone))]
1192 pub async fn download_file_stream(
1193 &self,
1194 #[builder(into)]
1196 filename: String,
1197 #[builder(into)]
1199 revision: Option<String>,
1200 range: Option<std::ops::Range<u64>>,
1205 #[builder(into)]
1208 progress: Option<Progress>,
1209 ) -> HFResult<(Option<u64>, HFByteStream)> {
1210 Box::pin(self.download_file_stream_impl(DownloadFileStreamParams {
1211 filename,
1212 revision,
1213 range,
1214 progress,
1215 }))
1216 .await
1217 }
1218
1219 #[builder(finish_fn = send, derive(Debug, Clone))]
1236 pub async fn download_file_to_bytes(
1237 &self,
1238 #[builder(into)]
1240 filename: String,
1241 #[builder(into)]
1243 revision: Option<String>,
1244 range: Option<std::ops::Range<u64>>,
1249 #[builder(into)]
1252 progress: Option<Progress>,
1253 ) -> HFResult<bytes::Bytes> {
1254 Box::pin(self.download_file_to_bytes_impl(DownloadFileStreamParams {
1255 filename,
1256 revision,
1257 range,
1258 progress,
1259 }))
1260 .await
1261 }
1262
1263 #[cfg(not(target_family = "wasm"))]
1286 #[builder(finish_fn = send, derive(Debug, Clone))]
1287 pub async fn snapshot_download(
1288 &self,
1289 #[builder(into)]
1291 revision: Option<String>,
1292 allow_patterns: Option<Vec<String>>,
1295 ignore_patterns: Option<Vec<String>>,
1297 #[builder(into)]
1299 local_dir: Option<PathBuf>,
1300 #[builder(default)]
1302 force_download: bool,
1303 #[builder(default)]
1305 local_files_only: bool,
1306 max_workers: Option<usize>,
1308 #[builder(into)]
1310 progress: Option<Progress>,
1311 ) -> HFResult<PathBuf> {
1312 Box::pin(self.snapshot_download_impl(SnapshotDownloadParams {
1313 revision,
1314 allow_patterns,
1315 ignore_patterns,
1316 local_dir,
1317 force_download,
1318 local_files_only,
1319 max_workers,
1320 progress,
1321 }))
1322 .await
1323 }
1324}
1325
1326#[cfg(feature = "blocking")]
1327#[bon]
1328impl<T: RepoType> crate::blocking::HFRepositorySync<T> {
1329 #[builder(finish_fn = send, derive(Debug, Clone))]
1332 pub fn download_file(
1333 &self,
1334 #[builder(into)] filename: String,
1335 #[builder(into)] local_dir: Option<PathBuf>,
1336 #[builder(into)] revision: Option<String>,
1337 #[builder(default)] force_download: bool,
1338 #[builder(default)] local_files_only: bool,
1339 #[builder(into)] progress: Option<Progress>,
1340 ) -> HFResult<PathBuf> {
1341 self.runtime.block_on(
1342 self.inner
1343 .download_file()
1344 .filename(filename)
1345 .maybe_local_dir(local_dir)
1346 .maybe_revision(revision)
1347 .force_download(force_download)
1348 .local_files_only(local_files_only)
1349 .maybe_progress(progress)
1350 .send(),
1351 )
1352 }
1353
1354 #[builder(finish_fn = send, derive(Debug, Clone))]
1357 pub fn download_file_to_bytes(
1358 &self,
1359 #[builder(into)] filename: String,
1360 #[builder(into)] revision: Option<String>,
1361 range: Option<std::ops::Range<u64>>,
1362 #[builder(into)] progress: Option<Progress>,
1363 ) -> HFResult<bytes::Bytes> {
1364 self.runtime.block_on(
1365 self.inner
1366 .download_file_to_bytes()
1367 .filename(filename)
1368 .maybe_revision(revision)
1369 .maybe_range(range)
1370 .maybe_progress(progress)
1371 .send(),
1372 )
1373 }
1374
1375 #[builder(finish_fn = send, derive(Debug, Clone))]
1378 pub fn snapshot_download(
1379 &self,
1380 #[builder(into)] revision: Option<String>,
1381 allow_patterns: Option<Vec<String>>,
1382 ignore_patterns: Option<Vec<String>>,
1383 #[builder(into)] local_dir: Option<PathBuf>,
1384 #[builder(default)] force_download: bool,
1385 #[builder(default)] local_files_only: bool,
1386 max_workers: Option<usize>,
1387 #[builder(into)] progress: Option<Progress>,
1388 ) -> HFResult<PathBuf> {
1389 self.runtime.block_on(
1390 self.inner
1391 .snapshot_download()
1392 .maybe_revision(revision)
1393 .maybe_allow_patterns(allow_patterns)
1394 .maybe_ignore_patterns(ignore_patterns)
1395 .maybe_local_dir(local_dir)
1396 .force_download(force_download)
1397 .local_files_only(local_files_only)
1398 .maybe_max_workers(max_workers)
1399 .maybe_progress(progress)
1400 .send(),
1401 )
1402 }
1403}