汉秋
d7fec35715
将本地存储与上传编排下沉到 Features/TravelAlbum，删除 USB/PTP 相关代码与测试。 Co-authored-by: Cursor <cursoragent@cursor.com>
246 lines
9.7 KiB
Swift
246 lines
9.7 KiB
Swift
//
|
||
// WiredCameraTransferPipeline.swift
|
||
// suixinkan
|
||
//
|
||
|
||
import Foundation
|
||
|
||
/// 有线传图上传编排器，负责管理本地照片的上传任务（不含 USB/OTG 相机连接）。
|
||
@MainActor
|
||
final class WiredCameraTransferPipeline {
|
||
var onTasksUpdated: (([CameraTransferTask]) -> Void)?
|
||
|
||
private(set) var tasks: [CameraTransferTask] = []
|
||
private var uploadSink: (any CameraAssetUploadSink)?
|
||
private var uploadEnabled = true
|
||
private var downloadScope = CameraDownloadStorage.Scope(accountKey: "unscoped", albumID: 0)
|
||
private var activeUploadAssetIDs: Set<String> = []
|
||
private var autoUploadEligibleAssetIDs: Set<String> = []
|
||
private var pendingDeferredNotifyTask: Task<Void, Never>?
|
||
private var lastDeferredNotifyDate = Date.distantPast
|
||
private var lastNotifiedProgressByAssetID: [String: Int] = [:]
|
||
private let maxRetries = 3
|
||
private let deferredNotifyInterval: TimeInterval = 0.15
|
||
private let maxConcurrentUploads = 3
|
||
private var isUIAttached = false
|
||
/// 用户已从列表删除、不应再展示的 asset ID。
|
||
private(set) var excludedAssetIDs: Set<String> = []
|
||
|
||
/// 绑定 UI 回调。
|
||
func attachToUI(onTasksUpdated: @escaping ([CameraTransferTask]) -> Void) {
|
||
isUIAttached = true
|
||
self.onTasksUpdated = onTasksUpdated
|
||
}
|
||
|
||
/// 页面离开：取消 UI 回调与进行中的上传通知。
|
||
func detachFromUI() {
|
||
guard isUIAttached else { return }
|
||
isUIAttached = false
|
||
pendingDeferredNotifyTask?.cancel()
|
||
pendingDeferredNotifyTask = nil
|
||
activeUploadAssetIDs.removeAll()
|
||
onTasksUpdated = nil
|
||
}
|
||
|
||
/// 配置上传 Sink、自动上传开关与下载作用域。
|
||
func configure(
|
||
uploadSink: (any CameraAssetUploadSink)?,
|
||
uploadEnabled: Bool,
|
||
downloadScope: CameraDownloadStorage.Scope? = nil
|
||
) {
|
||
self.uploadSink = uploadSink
|
||
self.uploadEnabled = uploadEnabled
|
||
if let downloadScope {
|
||
self.downloadScope = downloadScope
|
||
}
|
||
}
|
||
|
||
/// 重试所有失败的上传任务。
|
||
func retryFailedUploads() async {
|
||
guard uploadEnabled, uploadSink != nil else { return }
|
||
let failedAssetIDs = tasks.filter { $0.status == .failed }.map(\.assetID)
|
||
for index in tasks.indices where tasks[index].status == .failed {
|
||
tasks[index].status = .downloaded
|
||
tasks[index].errorMessage = nil
|
||
}
|
||
notify()
|
||
await uploadAssets(withIDs: failedAssetIDs)
|
||
}
|
||
|
||
/// 手动上传指定 asset 列表。
|
||
func uploadAssets(withIDs assetIDs: [String]) async {
|
||
guard uploadSink != nil else { return }
|
||
var seenAssetIDs: Set<String> = []
|
||
let uniqueAssetIDs = assetIDs.filter { seenAssetIDs.insert($0).inserted }
|
||
var currentIndex = 0
|
||
while currentIndex < uniqueAssetIDs.count {
|
||
let upperBound = min(currentIndex + maxConcurrentUploads, uniqueAssetIDs.count)
|
||
let batch = uniqueAssetIDs[currentIndex ..< upperBound]
|
||
let uploadTasks = batch.map { assetID in
|
||
Task { @MainActor in
|
||
await self.uploadAssetIfNeeded(assetID: assetID)
|
||
}
|
||
}
|
||
for task in uploadTasks {
|
||
await task.value
|
||
}
|
||
currentIndex = upperBound
|
||
}
|
||
}
|
||
|
||
/// 重传单张照片；若管道中无任务则根据本地相对路径恢复。
|
||
func retryUpload(assetID: String, filename: String, localPath: String) async {
|
||
guard uploadSink != nil else { return }
|
||
guard !activeUploadAssetIDs.contains(assetID) else { return }
|
||
|
||
if let index = tasks.firstIndex(where: { $0.assetID == assetID }) {
|
||
if tasks[index].status == .uploaded { return }
|
||
if tasks[index].localPath?.isEmpty != false, !localPath.isEmpty {
|
||
tasks[index].localPath = localPath
|
||
}
|
||
tasks[index].status = .downloaded
|
||
tasks[index].errorMessage = nil
|
||
} else {
|
||
guard !localPath.isEmpty,
|
||
let localURL = CameraDownloadStorage.resolveLocalURL(from: localPath),
|
||
FileManager.default.fileExists(atPath: localURL.path) else {
|
||
return
|
||
}
|
||
tasks.insert(
|
||
CameraTransferTask(
|
||
assetID: assetID,
|
||
filename: filename,
|
||
localPath: localPath,
|
||
capturedAt: CameraTransferTask.capturedAtString(forLocalPath: localPath),
|
||
status: .downloaded
|
||
),
|
||
at: 0
|
||
)
|
||
}
|
||
|
||
notify()
|
||
await uploadTask(assetID: assetID)
|
||
}
|
||
|
||
/// 按 assetID 查找任务。
|
||
func task(forAssetID assetID: String) -> CameraTransferTask? {
|
||
tasks.first { $0.assetID == assetID }
|
||
}
|
||
|
||
/// 同步用户已删除的 asset ID，防止再次带回。
|
||
func setExcludedAssetIDs(_ ids: Set<String>) {
|
||
excludedAssetIDs = ids
|
||
}
|
||
|
||
/// 从管道移除 asset。
|
||
func removeAsset(assetID: String) {
|
||
guard !assetID.isEmpty else { return }
|
||
excludedAssetIDs.insert(assetID)
|
||
tasks.removeAll { $0.assetID == assetID }
|
||
autoUploadEligibleAssetIDs.remove(assetID)
|
||
activeUploadAssetIDs.remove(assetID)
|
||
lastNotifiedProgressByAssetID.removeValue(forKey: assetID)
|
||
}
|
||
|
||
private func uploadAssetIfNeeded(assetID: String) async {
|
||
if activeUploadAssetIDs.contains(assetID) { return }
|
||
guard let index = tasks.firstIndex(where: { $0.assetID == assetID }) else { return }
|
||
if tasks[index].status == .uploaded { return }
|
||
|
||
activeUploadAssetIDs.insert(assetID)
|
||
defer { activeUploadAssetIDs.remove(assetID) }
|
||
|
||
guard let refreshedIndex = tasks.firstIndex(where: { $0.assetID == assetID }) else { return }
|
||
if tasks[refreshedIndex].status == .downloaded || tasks[refreshedIndex].status == .failed {
|
||
tasks[refreshedIndex].status = .downloaded
|
||
tasks[refreshedIndex].errorMessage = nil
|
||
await uploadTask(assetID: assetID)
|
||
}
|
||
}
|
||
|
||
private func uploadTask(assetID: String) async {
|
||
guard let index = tasks.firstIndex(where: { $0.assetID == assetID }),
|
||
let sink = uploadSink,
|
||
let localURL = tasks[index].localURL
|
||
else { return }
|
||
|
||
tasks[index].status = .uploading
|
||
tasks[index].errorMessage = nil
|
||
notify()
|
||
|
||
let filename = tasks[index].filename
|
||
let fileType = filename.lowercased().hasSuffix(".mp4") ? 1 : 2
|
||
var attempt = 0
|
||
while attempt < maxRetries {
|
||
do {
|
||
let remoteURL = try await sink.upload(
|
||
localURL: localURL,
|
||
fileName: filename,
|
||
fileType: fileType,
|
||
progress: { [weak self] progress in
|
||
Task { @MainActor in
|
||
guard let self,
|
||
let taskIndex = self.tasks.firstIndex(where: { $0.assetID == assetID }) else { return }
|
||
let lastNotifiedProgress = self.lastNotifiedProgressByAssetID[assetID] ?? -1
|
||
guard progress == 100 || progress - lastNotifiedProgress >= 10 else { return }
|
||
self.lastNotifiedProgressByAssetID[assetID] = progress
|
||
self.tasks[taskIndex].progress = progress
|
||
self.notify(progress == 100 ? .immediate : .deferred)
|
||
}
|
||
}
|
||
)
|
||
guard let taskIndex = tasks.firstIndex(where: { $0.assetID == assetID }) else { return }
|
||
tasks[taskIndex].status = .uploaded
|
||
tasks[taskIndex].remoteURL = remoteURL
|
||
tasks[taskIndex].progress = 100
|
||
autoUploadEligibleAssetIDs.remove(assetID)
|
||
lastNotifiedProgressByAssetID.removeValue(forKey: assetID)
|
||
notify()
|
||
return
|
||
} catch {
|
||
attempt += 1
|
||
guard let taskIndex = tasks.firstIndex(where: { $0.assetID == assetID }) else { return }
|
||
if attempt >= maxRetries {
|
||
tasks[taskIndex].status = .failed
|
||
tasks[taskIndex].errorMessage = error.localizedDescription
|
||
lastNotifiedProgressByAssetID.removeValue(forKey: assetID)
|
||
notify()
|
||
} else {
|
||
let delay = UInt64(pow(2.0, Double(attempt)))
|
||
try? await Task.sleep(nanoseconds: delay * 1_000_000_000)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
private func notify(_ mode: WiredTransferNotifyMode = .immediate) {
|
||
switch mode {
|
||
case .immediate:
|
||
pendingDeferredNotifyTask?.cancel()
|
||
pendingDeferredNotifyTask = nil
|
||
onTasksUpdated?(tasks)
|
||
case .deferred:
|
||
let elapsed = Date().timeIntervalSince(lastDeferredNotifyDate)
|
||
if elapsed >= deferredNotifyInterval {
|
||
lastDeferredNotifyDate = Date()
|
||
onTasksUpdated?(tasks)
|
||
return
|
||
}
|
||
guard pendingDeferredNotifyTask == nil else { return }
|
||
let delay = UInt64((deferredNotifyInterval - elapsed) * 1_000_000_000)
|
||
pendingDeferredNotifyTask = Task { @MainActor in
|
||
try? await Task.sleep(nanoseconds: delay)
|
||
guard !Task.isCancelled else { return }
|
||
self.lastDeferredNotifyDate = Date()
|
||
self.pendingDeferredNotifyTask = nil
|
||
self.onTasksUpdated?(self.tasks)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
private enum WiredTransferNotifyMode {
|
||
case immediate
|
||
case deferred
|
||
}
|