439 lines
24 KiB
C#
439 lines
24 KiB
C#
using System.Text;
|
||
using dy.net.model.dto;
|
||
using dy.net.model.entity;
|
||
using dy.net.storage;
|
||
using dy.net.utils;
|
||
using Newtonsoft.Json;
|
||
using SqlSugar;
|
||
using MediaStorageType = dy.net.model.dto.StorageType;
|
||
|
||
namespace dy.net.service
|
||
{
|
||
public class StorageMigrationItemProcessor
|
||
{
|
||
private readonly ISqlSugarClient _db;
|
||
private readonly OpenListSettingsService _settings;
|
||
private readonly OpenListMediaStorage _openList;
|
||
private readonly DouyinHttpClientService _http;
|
||
private readonly DouyinMigrationSourceResolver _sourceResolver;
|
||
|
||
public StorageMigrationItemProcessor(
|
||
ISqlSugarClient db,
|
||
OpenListSettingsService settings,
|
||
OpenListMediaStorage openList,
|
||
DouyinHttpClientService http,
|
||
DouyinMigrationSourceResolver sourceResolver)
|
||
{
|
||
_db = db;
|
||
_settings = settings;
|
||
_openList = openList;
|
||
_http = http;
|
||
_sourceResolver = sourceResolver;
|
||
}
|
||
|
||
public async Task ProcessAsync(string itemId, CancellationToken cancellationToken)
|
||
{
|
||
var item = await _db.Queryable<StorageMigrationItem>().InSingleAsync(itemId);
|
||
if (item == null || item.Stage != StorageMigrationItemStage.Pending) return;
|
||
var task = await _db.Queryable<StorageMigrationTask>().InSingleAsync(item.TaskId);
|
||
if (task?.Status != StorageMigrationTaskStatus.Running) return;
|
||
if (task.TargetStorageType != MediaStorageType.OpenList)
|
||
{
|
||
await _db.Updateable<StorageMigrationTask>()
|
||
.SetColumns(x => new StorageMigrationTask
|
||
{
|
||
Status = StorageMigrationTaskStatus.Cancelled,
|
||
ErrorMessage = "旧 WebDAV 迁移任务仅保留审计信息，不能继续写入。",
|
||
CompletedAt = DateTime.Now,
|
||
UpdatedAt = DateTime.Now
|
||
}).Where(x => x.Id == task.Id).ExecuteCommandAsync();
|
||
return;
|
||
}
|
||
|
||
var targetSettings = await _settings.GetAsync();
|
||
var fingerprint = StorageConfigurationFingerprint.Create(targetSettings);
|
||
if (!string.Equals(fingerprint, task.ConfigurationFingerprint, StringComparison.Ordinal))
|
||
{
|
||
await _db.Updateable<StorageMigrationTask>()
|
||
.SetColumns(x => new StorageMigrationTask
|
||
{
|
||
Status = StorageMigrationTaskStatus.Paused,
|
||
ErrorMessage = "OpenList 配置已变化，请恢复原配置并重新预检。",
|
||
UpdatedAt = DateTime.Now
|
||
}).Where(x => x.Id == task.Id).ExecuteCommandAsync();
|
||
return;
|
||
}
|
||
var targetStorage = _openList.Bind(targetSettings);
|
||
|
||
var snapshot = JsonConvert.DeserializeObject<DouyinVideo>(item.OldSnapshotJson)
|
||
?? throw new InvalidOperationException("迁移条目缺少旧视频快照");
|
||
var cookie = await _db.Queryable<DouyinCookie>().InSingleAsync(snapshot.CookieId);
|
||
var roots = StorageMigrationPathPolicy.GetLocalRoots(cookie);
|
||
var attachments = JsonConvert.DeserializeObject<List<StorageMigrationService.MigrationAttachmentPlan>>(item.AttachmentPlanJson) ?? new();
|
||
var uploaded = DeserializeSet(item.UploadedPathsJson);
|
||
var owned = DeserializeSet(item.OwnedRemotePathsJson);
|
||
var warnings = new List<string>();
|
||
|
||
try
|
||
{
|
||
if (targetStorage is ICanonicalMediaStorage canonicalStorage)
|
||
{
|
||
item.TargetVideoPath = await canonicalStorage.CanonicalizePathAsync(
|
||
item.TargetVideoPath, true, cancellationToken);
|
||
foreach (var attachment in attachments.Where(x => !string.IsNullOrWhiteSpace(x.TargetPath)))
|
||
attachment.TargetPath = await canonicalStorage.CanonicalizePathAsync(
|
||
attachment.TargetPath, true, cancellationToken);
|
||
item.AttachmentPlanJson = JsonConvert.SerializeObject(attachments);
|
||
item.UpdatedAt = DateTime.Now;
|
||
await _db.Updateable(item).ExecuteCommandAsync();
|
||
}
|
||
if (item.AdoptExistingTarget && snapshot.StorageType == MediaStorageType.WebDav)
|
||
{
|
||
await AdoptExistingAsync(item, snapshot, targetStorage, cancellationToken);
|
||
return;
|
||
}
|
||
item.Stage = StorageMigrationItemStage.Uploading;
|
||
item.Attempts++;
|
||
item.ErrorMessage = null;
|
||
item.UpdatedAt = DateTime.Now;
|
||
await _db.Updateable(item).ExecuteCommandAsync();
|
||
|
||
var mainLength = await UploadMainAsync(snapshot, cookie, roots, item, targetStorage, uploaded, owned, cancellationToken);
|
||
item.ExpectedLength = mainLength;
|
||
|
||
var newDynamic = new List<DouyinMergeVideoDto>();
|
||
foreach (var attachment in attachments)
|
||
{
|
||
if (attachment.TargetPath == item.TargetVideoPath) continue;
|
||
try
|
||
{
|
||
var length = await UploadAttachmentAsync(attachment, cookie, roots, targetStorage, uploaded, owned, cancellationToken);
|
||
if (length.HasValue && attachment.Kind == "dynamic")
|
||
newDynamic.Add(new DouyinMergeVideoDto { Path = attachment.TargetPath, Height = attachment.Height, Width = attachment.Width });
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
warnings.Add($"{AttachmentName(attachment.Kind)}：{ex.Message}");
|
||
}
|
||
}
|
||
|
||
item.Stage = StorageMigrationItemStage.Verifying;
|
||
item.UploadedPathsJson = JsonConvert.SerializeObject(uploaded);
|
||
item.OwnedRemotePathsJson = JsonConvert.SerializeObject(owned);
|
||
item.WarningMessage = warnings.Count == 0 ? null : string.Join("；", warnings);
|
||
item.UpdatedAt = DateTime.Now;
|
||
await _db.Updateable(item).ExecuteCommandAsync();
|
||
|
||
var remoteLength = await targetStorage.GetLengthAsync(item.TargetVideoPath, cancellationToken);
|
||
if (!remoteLength.HasValue || remoteLength.Value != mainLength)
|
||
throw new IOException($"远端主媒体长度校验失败，期望 {mainLength}，实际 {remoteLength?.ToString() ?? "不存在"}");
|
||
|
||
var migrated = JsonConvert.DeserializeObject<DouyinVideo>(item.OldSnapshotJson);
|
||
migrated.StorageType = MediaStorageType.OpenList;
|
||
migrated.VideoSavePath = item.TargetVideoPath;
|
||
migrated.VideoCoverSavePath = attachments.FirstOrDefault(x => x.Kind == "cover" && uploaded.Contains(x.TargetPath))?.TargetPath;
|
||
migrated.AuthorAvatar = attachments.FirstOrDefault(x => x.Kind == "avatar" && uploaded.Contains(x.TargetPath))?.TargetPath;
|
||
migrated.DynamicVideos = newDynamic.Count == 0 ? null : JsonConvert.SerializeObject(newDynamic);
|
||
migrated.FileSize = mainLength;
|
||
var category = string.IsNullOrWhiteSpace(snapshot.CateId) ? null : await _db.Queryable<DouyinCollectCate>().InSingleAsync(snapshot.CateId);
|
||
foreach (var nfo in NfoContentBuilder.Build(migrated, category?.Name))
|
||
{
|
||
var bytes = Encoding.UTF8.GetBytes(nfo.Value);
|
||
var existed = await targetStorage.ExistsAsync(nfo.Key, cancellationToken);
|
||
await using var content = new MemoryStream(bytes, false);
|
||
await targetStorage.WriteAsync(nfo.Key, content, bytes.Length, "application/xml", cancellationToken);
|
||
uploaded.Add(nfo.Key);
|
||
if (!existed) owned.Add(nfo.Key);
|
||
}
|
||
|
||
item.Stage = StorageMigrationItemStage.Committing;
|
||
item.UploadedPathsJson = JsonConvert.SerializeObject(uploaded);
|
||
item.OwnedRemotePathsJson = JsonConvert.SerializeObject(owned);
|
||
item.UpdatedAt = DateTime.Now;
|
||
await _db.Updateable(item).ExecuteCommandAsync();
|
||
|
||
var transaction = await _db.Ado.UseTranAsync(async () =>
|
||
{
|
||
var current = await _db.Queryable<DouyinVideo>().InSingleAsync(item.VideoId)
|
||
?? throw new InvalidOperationException("原视频记录已不存在");
|
||
if (current.StorageType == MediaStorageType.OpenList && current.VideoSavePath == item.TargetVideoPath)
|
||
{
|
||
// A previous process committed the video and died before updating the item.
|
||
}
|
||
else if (await TryReconcileCurrentOpenListAsync(
|
||
current, item, snapshot, targetStorage, cancellationToken))
|
||
{
|
||
// A normal sync already converged this record while migration was running.
|
||
// The verified OpenList record is authoritative and must not be overwritten.
|
||
}
|
||
else
|
||
{
|
||
if (current.StorageType != snapshot.StorageType || current.VideoSavePath != snapshot.VideoSavePath)
|
||
throw new InvalidOperationException("原视频记录在迁移期间已变化，拒绝覆盖");
|
||
current.StorageType = MediaStorageType.OpenList;
|
||
current.VideoSavePath = migrated.VideoSavePath;
|
||
current.VideoCoverSavePath = migrated.VideoCoverSavePath;
|
||
current.AuthorAvatar = migrated.AuthorAvatar;
|
||
current.DynamicVideos = migrated.DynamicVideos;
|
||
current.FileSize = migrated.FileSize;
|
||
var changed = await _db.Updateable(current).ExecuteCommandAsync();
|
||
if (changed != 1) throw new InvalidOperationException("提交原视频记录失败");
|
||
}
|
||
|
||
item.Stage = warnings.Count == 0 ? StorageMigrationItemStage.Succeeded : StorageMigrationItemStage.SucceededWithWarnings;
|
||
item.CompletedAt = DateTime.Now;
|
||
item.UpdatedAt = DateTime.Now;
|
||
item.WarningMessage = warnings.Count == 0 ? null : string.Join("；", warnings);
|
||
var itemChanged = await _db.Updateable(item).ExecuteCommandAsync();
|
||
if (itemChanged != 1) throw new InvalidOperationException("提交迁移条目失败");
|
||
});
|
||
if (!transaction.IsSuccess) throw transaction.ErrorException ?? new InvalidOperationException("数据库事务提交失败");
|
||
}
|
||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||
{
|
||
await FailAsync(item, "应用正在停止，条目将在下次启动后重试。", resetToPending: true);
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
Serilog.Log.Warning(ex, "存储迁移条目失败：{ItemId}/{VideoId}", item.Id, item.VideoId);
|
||
item.UploadedPathsJson = JsonConvert.SerializeObject(uploaded);
|
||
item.OwnedRemotePathsJson = JsonConvert.SerializeObject(owned);
|
||
await FailAsync(item, ex.Message, resetToPending: false);
|
||
}
|
||
}
|
||
|
||
private async Task AdoptExistingAsync(
|
||
StorageMigrationItem item,
|
||
DouyinVideo snapshot,
|
||
IMediaStorage targetStorage,
|
||
CancellationToken cancellationToken)
|
||
{
|
||
item.Stage = StorageMigrationItemStage.Verifying;
|
||
item.Attempts++;
|
||
item.ErrorMessage = null;
|
||
item.UpdatedAt = DateTime.Now;
|
||
await _db.Updateable(item).ExecuteCommandAsync();
|
||
|
||
var length = await targetStorage.GetLengthAsync(item.TargetVideoPath, cancellationToken);
|
||
if (!length.HasValue || length <= 0
|
||
|| snapshot.FileSize > 0 && length.Value != snapshot.FileSize)
|
||
throw new IOException($"OpenList 接管校验失败：期望 {snapshot.FileSize}，实际 {length?.ToString() ?? "不存在"}");
|
||
|
||
var warnings = new List<string>();
|
||
foreach (var attachment in new[]
|
||
{
|
||
(Name: "封面", Path: snapshot.VideoCoverSavePath),
|
||
(Name: "头像", Path: snapshot.AuthorAvatar)
|
||
})
|
||
{
|
||
if (string.IsNullOrWhiteSpace(attachment.Path)) continue;
|
||
try
|
||
{
|
||
if (!await targetStorage.ExistsAsync(attachment.Path, cancellationToken))
|
||
warnings.Add($"{attachment.Name}在 OpenList 中不存在");
|
||
}
|
||
catch (Exception ex) { warnings.Add($"{attachment.Name}检查失败：{ex.GetBaseException().Message}"); }
|
||
}
|
||
|
||
item.Stage = StorageMigrationItemStage.Committing;
|
||
item.ExpectedLength = length.Value;
|
||
item.WarningMessage = warnings.Count == 0 ? null : string.Join("；", warnings);
|
||
item.UpdatedAt = DateTime.Now;
|
||
await _db.Updateable(item).ExecuteCommandAsync();
|
||
|
||
var transaction = await _db.Ado.UseTranAsync(async () =>
|
||
{
|
||
var current = await _db.Queryable<DouyinVideo>().InSingleAsync(item.VideoId)
|
||
?? throw new InvalidOperationException("原视频记录已不存在");
|
||
if (current.StorageType == MediaStorageType.OpenList
|
||
&& current.VideoSavePath == item.TargetVideoPath)
|
||
{
|
||
// 上次进程可能已提交视频，但尚未更新迁移条目。
|
||
}
|
||
else if (await TryReconcileCurrentOpenListAsync(
|
||
current, item, snapshot, targetStorage, cancellationToken))
|
||
{
|
||
// 普通同步已先完成接管；验证当前远端文件后直接收敛迁移条目。
|
||
}
|
||
else
|
||
{
|
||
if (current.StorageType != snapshot.StorageType
|
||
|| current.VideoSavePath != snapshot.VideoSavePath)
|
||
throw new InvalidOperationException("原视频记录在接管期间已变化，拒绝覆盖");
|
||
current.StorageType = MediaStorageType.OpenList;
|
||
current.VideoSavePath = item.TargetVideoPath;
|
||
current.FileSize = length.Value;
|
||
if (await _db.Updateable(current).ExecuteCommandAsync() != 1)
|
||
throw new InvalidOperationException("提交 OpenList 接管记录失败");
|
||
}
|
||
item.Stage = warnings.Count == 0
|
||
? StorageMigrationItemStage.Succeeded
|
||
: StorageMigrationItemStage.SucceededWithWarnings;
|
||
item.CompletedAt = DateTime.Now;
|
||
item.UpdatedAt = DateTime.Now;
|
||
if (await _db.Updateable(item).ExecuteCommandAsync() != 1)
|
||
throw new InvalidOperationException("提交 OpenList 接管条目失败");
|
||
});
|
||
if (!transaction.IsSuccess)
|
||
throw transaction.ErrorException ?? new InvalidOperationException("OpenList 接管事务提交失败");
|
||
}
|
||
|
||
private static async Task<bool> TryReconcileCurrentOpenListAsync(
|
||
DouyinVideo current,
|
||
StorageMigrationItem item,
|
||
DouyinVideo snapshot,
|
||
IMediaStorage targetStorage,
|
||
CancellationToken cancellationToken)
|
||
{
|
||
if (current?.StorageType != MediaStorageType.OpenList
|
||
|| string.IsNullOrWhiteSpace(current.VideoSavePath)) return false;
|
||
|
||
var canonicalPath = targetStorage is ICanonicalMediaStorage canonicalStorage
|
||
? await canonicalStorage.CanonicalizePathAsync(current.VideoSavePath, false, cancellationToken)
|
||
: current.VideoSavePath;
|
||
var remoteLength = await targetStorage.GetLengthAsync(canonicalPath, cancellationToken);
|
||
if (!remoteLength.HasValue || remoteLength.Value <= 0)
|
||
throw new IOException($"并发接管记录已指向 OpenList，但远端主媒体不存在：{canonicalPath}");
|
||
|
||
var expectedLength = current.FileSize > 0
|
||
? current.FileSize
|
||
: item.ExpectedLength > 0 ? item.ExpectedLength : snapshot?.FileSize ?? 0;
|
||
if (expectedLength > 0 && remoteLength.Value != expectedLength)
|
||
throw new MediaIntegrityException(
|
||
$"并发接管后的远端主媒体长度不一致，期望 {expectedLength}，实际 {remoteLength.Value}");
|
||
|
||
item.TargetVideoPath = canonicalPath;
|
||
item.ExpectedLength = remoteLength.Value;
|
||
return true;
|
||
}
|
||
|
||
private async Task<long> UploadMainAsync(
|
||
DouyinVideo snapshot,
|
||
DouyinCookie cookie,
|
||
IReadOnlyList<string> roots,
|
||
StorageMigrationItem item,
|
||
IMediaStorage targetStorage,
|
||
HashSet<string> uploaded,
|
||
HashSet<string> owned,
|
||
CancellationToken cancellationToken)
|
||
{
|
||
if (SafeLocalMigrationFile.TryResolve(snapshot.VideoSavePath, roots, out var local, out _)
|
||
&& new FileInfo(local).Length > 0)
|
||
{
|
||
var expected = new FileInfo(local).Length;
|
||
await UploadLocalAsync(local, item.TargetVideoPath, expected, targetStorage, uploaded, owned, cancellationToken);
|
||
return expected;
|
||
}
|
||
|
||
var knownLength = item.ExpectedLength > 0 ? item.ExpectedLength : snapshot.FileSize;
|
||
var existingLength = await targetStorage.GetLengthAsync(item.TargetVideoPath, cancellationToken);
|
||
if (knownLength > 0 && existingLength == knownLength)
|
||
{
|
||
uploaded.Add(item.TargetVideoPath);
|
||
return knownLength;
|
||
}
|
||
|
||
var existed = existingLength.HasValue;
|
||
if (!string.IsNullOrWhiteSpace(snapshot.VideoUrl))
|
||
{
|
||
var direct = await _http.DownloadToStorageAsync(MediaStorageType.OpenList, snapshot.VideoUrl, item.TargetVideoPath,
|
||
cookie?.Cookies, cancellationToken: cancellationToken, maxRetryCount: 1, storageOverride: targetStorage);
|
||
if (direct.Success)
|
||
{
|
||
var directLength = await targetStorage.GetLengthAsync(item.TargetVideoPath, cancellationToken);
|
||
if (directLength.HasValue && directLength.Value > 0)
|
||
{
|
||
uploaded.Add(item.TargetVideoPath);
|
||
if (!existed) owned.Add(item.TargetVideoPath);
|
||
return directLength.Value;
|
||
}
|
||
}
|
||
}
|
||
|
||
var urls = (await _sourceResolver.ResolveAsync(snapshot, cookie, cancellationToken)).ToList();
|
||
if (urls.Count == 0) throw new FileNotFoundException("本地主媒体缺失，记录 URL 和抖音全量列表均未找到可用回源地址");
|
||
var result = await _http.DownloadToStorageAsync(MediaStorageType.OpenList, urls[0], item.TargetVideoPath,
|
||
cookie?.Cookies, urls.Skip(1).ToList(), cancellationToken, Math.Min(6, urls.Count), targetStorage);
|
||
if (!result.Success) throw result.ToException();
|
||
var length = await targetStorage.GetLengthAsync(item.TargetVideoPath, cancellationToken);
|
||
if (!length.HasValue || length.Value <= 0) throw new IOException("回源上传后远端主媒体为空");
|
||
uploaded.Add(item.TargetVideoPath);
|
||
if (!existed) owned.Add(item.TargetVideoPath);
|
||
return length.Value;
|
||
}
|
||
|
||
private async Task<long?> UploadAttachmentAsync(
|
||
StorageMigrationService.MigrationAttachmentPlan attachment,
|
||
DouyinCookie cookie,
|
||
IReadOnlyList<string> roots,
|
||
IMediaStorage targetStorage,
|
||
HashSet<string> uploaded,
|
||
HashSet<string> owned,
|
||
CancellationToken cancellationToken)
|
||
{
|
||
if (SafeLocalMigrationFile.TryResolve(attachment.SourcePath, roots, out var local, out _))
|
||
{
|
||
var length = new FileInfo(local).Length;
|
||
if (length <= 0) throw new IOException("本地附件为空");
|
||
await UploadLocalAsync(local, attachment.TargetPath, length, targetStorage, uploaded, owned, cancellationToken);
|
||
return length;
|
||
}
|
||
if (string.IsNullOrWhiteSpace(attachment.SourceUrl)) throw new FileNotFoundException("本地附件缺失且无回源地址");
|
||
var before = await targetStorage.GetLengthAsync(attachment.TargetPath, cancellationToken);
|
||
var result = await _http.DownloadToStorageAsync(MediaStorageType.OpenList, attachment.SourceUrl, attachment.TargetPath,
|
||
cookie?.Cookies, cancellationToken: cancellationToken, storageOverride: targetStorage);
|
||
if (!result.Success) throw result.ToException();
|
||
var after = await targetStorage.GetLengthAsync(attachment.TargetPath, cancellationToken);
|
||
if (!after.HasValue || after.Value <= 0) throw new IOException("附件上传后为空");
|
||
uploaded.Add(attachment.TargetPath);
|
||
if (!before.HasValue) owned.Add(attachment.TargetPath);
|
||
return after;
|
||
}
|
||
|
||
private async Task UploadLocalAsync(
|
||
string local,
|
||
string target,
|
||
long expected,
|
||
IMediaStorage targetStorage,
|
||
HashSet<string> uploaded,
|
||
HashSet<string> owned,
|
||
CancellationToken cancellationToken)
|
||
{
|
||
var before = await targetStorage.GetLengthAsync(target, cancellationToken);
|
||
if (before == expected)
|
||
{
|
||
uploaded.Add(target);
|
||
return;
|
||
}
|
||
await using var source = new FileStream(local, FileMode.Open, FileAccess.Read, FileShare.Read, 81920, true);
|
||
await targetStorage.WriteAsync(target, source, expected, null, cancellationToken);
|
||
var after = await targetStorage.GetLengthAsync(target, cancellationToken);
|
||
if (after != expected) throw new IOException($"附件长度校验失败，期望 {expected}，实际 {after?.ToString() ?? "不存在"}");
|
||
uploaded.Add(target);
|
||
if (!before.HasValue) owned.Add(target);
|
||
}
|
||
|
||
private async Task FailAsync(StorageMigrationItem item, string error, bool resetToPending)
|
||
{
|
||
item.Stage = resetToPending ? StorageMigrationItemStage.Pending : StorageMigrationItemStage.Failed;
|
||
item.ErrorMessage = error;
|
||
item.UpdatedAt = DateTime.Now;
|
||
await _db.Updateable(item).ExecuteCommandAsync();
|
||
}
|
||
|
||
private static HashSet<string> DeserializeSet(string json)
|
||
{
|
||
try { return new HashSet<string>(JsonConvert.DeserializeObject<List<string>>(json) ?? new(), StringComparer.Ordinal); }
|
||
catch { return new HashSet<string>(StringComparer.Ordinal); }
|
||
}
|
||
|
||
private static string AttachmentName(string kind) => kind switch
|
||
{
|
||
"cover" => "封面",
|
||
"avatar" => "头像",
|
||
"dynamic" => "图文/动态附件",
|
||
_ => "附件"
|
||
};
|
||
}
|
||
}
|