feat: add fnOS packaging, storage workflows and release pipeline
This commit is contained in:
@@ -0,0 +1,222 @@
|
||||
using dy.net.model.dto;
|
||||
using dy.net.model.entity;
|
||||
using dy.net.storage;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Newtonsoft.Json;
|
||||
using SqlSugar;
|
||||
|
||||
namespace dy.net.service
|
||||
{
|
||||
public class VideoTaskWorker : BackgroundService
|
||||
{
|
||||
private readonly IServiceScopeFactory _scopeFactory;
|
||||
|
||||
public VideoTaskWorker(IServiceScopeFactory scopeFactory) => _scopeFactory = scopeFactory;
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
using (var scope = _scopeFactory.CreateScope())
|
||||
{
|
||||
var tasks = scope.ServiceProvider.GetRequiredService<VideoTaskService>();
|
||||
await tasks.RecoverInterruptedAsync();
|
||||
await BackfillLegacyRedownloadsAsync(scope.ServiceProvider);
|
||||
await tasks.CleanupHistoryAsync();
|
||||
}
|
||||
var lastCleanup = DateTime.UtcNow;
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
try
|
||||
{
|
||||
var worked = await ProcessOneAsync(stoppingToken);
|
||||
if (DateTime.UtcNow - lastCleanup > TimeSpan.FromDays(1))
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
await scope.ServiceProvider.GetRequiredService<VideoTaskService>().CleanupHistoryAsync();
|
||||
lastCleanup = DateTime.UtcNow;
|
||||
}
|
||||
if (!worked) await Task.Delay(TimeSpan.FromSeconds(2), stoppingToken);
|
||||
}
|
||||
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { }
|
||||
catch (Exception ex)
|
||||
{
|
||||
Serilog.Log.Error(ex, "视频任务后台服务异常");
|
||||
await Task.Delay(TimeSpan.FromSeconds(3), stoppingToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<bool> ProcessOneAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>();
|
||||
var tasks = scope.ServiceProvider.GetRequiredService<VideoTaskService>();
|
||||
var candidates = await db.Queryable<VideoDownloadTaskItem>()
|
||||
.Where(x => x.WorkerManaged && (x.Stage == VideoTaskItemStage.Pending
|
||||
|| x.Stage == VideoTaskItemStage.WaitingForSource))
|
||||
.OrderBy(x => x.CreatedAt).Take(50).ToListAsync();
|
||||
VideoDownloadTaskItem item = null;
|
||||
var isSourceProbe = false;
|
||||
foreach (var candidate in candidates)
|
||||
{
|
||||
var access = candidate.Stage == VideoTaskItemStage.WaitingForSource
|
||||
? await tasks.TryAcquireSourceAsync(candidate.CookieId, candidate.Id)
|
||||
: await tasks.GetSourceAccessAsync(candidate.CookieId);
|
||||
if (access.Allowed)
|
||||
{
|
||||
item = candidate;
|
||||
isSourceProbe = access.IsProbe;
|
||||
if (item.Stage == VideoTaskItemStage.WaitingForSource)
|
||||
{
|
||||
item.Stage = VideoTaskItemStage.Pending;
|
||||
item.ErrorMessage = null;
|
||||
item.ErrorType = VideoTaskErrorType.None;
|
||||
item.RetryAfter = null;
|
||||
item.UpdatedAt = DateTime.Now;
|
||||
await db.Updateable(item).ExecuteCommandAsync();
|
||||
}
|
||||
break;
|
||||
}
|
||||
if (candidate.Stage == VideoTaskItemStage.Pending)
|
||||
await tasks.WaitItemForSourceAsync(candidate.Id, candidate.CookieId, access.Message);
|
||||
}
|
||||
if (item == null) return false;
|
||||
var task = await db.Queryable<VideoDownloadTask>().InSingleAsync(item.TaskId);
|
||||
if (task == null || task.Type == VideoTaskType.Redownload) return false;
|
||||
try
|
||||
{
|
||||
await tasks.EnsureTaskStorageAvailableAsync(task.Id);
|
||||
}
|
||||
catch (InvalidOperationException ex)
|
||||
{
|
||||
item.Stage = VideoTaskItemStage.WaitingForStorage;
|
||||
item.ErrorType = VideoTaskErrorType.StorageUnavailable;
|
||||
item.ErrorMessage = ex.Message;
|
||||
item.UpdatedAt = DateTime.Now;
|
||||
await db.Updateable(item).ExecuteCommandAsync();
|
||||
await tasks.BlockTaskForStorageAsync(task.Id, ex.Message);
|
||||
await tasks.RefreshCountsAsync(task.Id);
|
||||
return true;
|
||||
}
|
||||
await tasks.StartTaskAsync(task.Id);
|
||||
try
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(item.RetrySnapshotJson))
|
||||
throw new InvalidOperationException("该失败条目缺少可恢复快照，请重新执行对应的同步任务。");
|
||||
var video = JsonConvert.DeserializeObject<DouyinVideo>(item.RetrySnapshotJson)
|
||||
?? throw new InvalidOperationException("任务缺少视频恢复快照。");
|
||||
var cookie = await db.Queryable<DouyinCookie>().InSingleAsync(video.CookieId)
|
||||
?? throw new InvalidOperationException("任务所属抖音授权不存在。");
|
||||
if (string.IsNullOrWhiteSpace(cookie.Cookies)) throw new InvalidOperationException("Cookie 无效，无法恢复下载。");
|
||||
var storedUrls = DeserializeUrls(item.SourceUrlsJson);
|
||||
if (!string.IsNullOrWhiteSpace(video.VideoUrl) && !storedUrls.Contains(video.VideoUrl)) storedUrls.Insert(0, video.VideoUrl);
|
||||
var resolvedUrls = (await scope.ServiceProvider.GetRequiredService<DouyinMigrationSourceResolver>()
|
||||
.ResolveAsync(video, cookie, cancellationToken)).ToList();
|
||||
var urls = (isSourceProbe ? resolvedUrls.Concat(storedUrls) : storedUrls.Concat(resolvedUrls))
|
||||
.Where(x => !string.IsNullOrWhiteSpace(x)).Distinct(StringComparer.Ordinal).Take(12).ToList();
|
||||
if (urls.Count == 0) throw new InvalidOperationException("作品没有可用下载地址，可能已下架或设为私密。");
|
||||
await tasks.SetItemStageAsync(item.Id, VideoTaskItemStage.Downloading, video.VideoSavePath);
|
||||
var storage = scope.ServiceProvider.GetRequiredService<MediaStorageRouter>().Resolve(video.StorageType);
|
||||
var existingLength = await storage.GetLengthAsync(video.VideoSavePath, cancellationToken);
|
||||
if (!existingLength.HasValue || existingLength <= 0)
|
||||
{
|
||||
var downloaded = await scope.ServiceProvider.GetRequiredService<DouyinHttpClientService>()
|
||||
.DownloadToStorageAsync(video.StorageType, urls[0], video.VideoSavePath, cookie.Cookies,
|
||||
urls.Skip(1).ToList(), cancellationToken, Math.Min(12, urls.Count));
|
||||
if (!downloaded.Success)
|
||||
{
|
||||
await tasks.MarkDownloadFailedAsync(item.Id, cookie.Id, downloaded);
|
||||
await CompleteWhenIdleAsync(db, tasks, task.Id);
|
||||
return true;
|
||||
}
|
||||
video.VideoSavePath = downloaded.ActualSavePath;
|
||||
await tasks.RecordSourceSuccessAsync(cookie.Id);
|
||||
}
|
||||
await tasks.SetItemStageAsync(item.Id, VideoTaskItemStage.Verifying, video.VideoSavePath);
|
||||
var length = await storage.GetLengthAsync(video.VideoSavePath, cancellationToken) ?? 0;
|
||||
if (length <= 0) throw new IOException("下载后的主媒体不存在或为空。");
|
||||
if (item.ExpectedLength > 0 && length != item.ExpectedLength)
|
||||
throw new IOException($"下载长度不一致：期望 {item.ExpectedLength}，实际 {length}。");
|
||||
await tasks.SetItemStageAsync(item.Id, VideoTaskItemStage.Committing, video.VideoSavePath);
|
||||
video.FileSize = length;
|
||||
video.SyncTime = DateTime.Now;
|
||||
if (string.IsNullOrWhiteSpace(video.Id)) video.Id = Guid.NewGuid().ToString("N");
|
||||
await db.Storageable(video).ExecuteCommandAsync();
|
||||
var warning = "恢复任务仅保证主媒体，封面、头像或 NFO 将在后续同步补齐。";
|
||||
var cleanupPending = false;
|
||||
string cleanupError = null;
|
||||
if (video.SupersededStorageSnapshot != null)
|
||||
{
|
||||
try
|
||||
{
|
||||
await scope.ServiceProvider.GetRequiredService<DouyinVideoService>()
|
||||
.CleanupSupersededLocalArtifactsAsync(video.SupersededStorageSnapshot, video.VideoSavePath);
|
||||
}
|
||||
catch (Exception cleanupException)
|
||||
{
|
||||
cleanupPending = true;
|
||||
cleanupError = cleanupException.GetBaseException().Message;
|
||||
warning += $" 旧本地文件清理失败，可在确认远端文件后稍后重试：{cleanupError}";
|
||||
}
|
||||
}
|
||||
await tasks.MarkSucceededAsync(item.Id, length, video.Id, warning, cleanupPending, cleanupError);
|
||||
await CompleteWhenIdleAsync(db, tasks, task.Id);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
await tasks.MarkFailedAsync(item.Id, ex);
|
||||
await CompleteWhenIdleAsync(db, tasks, task.Id);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private static async Task CompleteWhenIdleAsync(ISqlSugarClient db, VideoTaskService tasks, string taskId)
|
||||
{
|
||||
var remaining = await db.Queryable<VideoDownloadTaskItem>().Where(x => x.TaskId == taskId
|
||||
&& (x.Stage == VideoTaskItemStage.Pending || x.Stage == VideoTaskItemStage.Downloading
|
||||
|| x.Stage == VideoTaskItemStage.Verifying || x.Stage == VideoTaskItemStage.Committing
|
||||
|| x.Stage == VideoTaskItemStage.WaitingForSource)).CountAsync();
|
||||
if (remaining == 0) await tasks.CompleteTaskAsync(taskId);
|
||||
}
|
||||
|
||||
private static List<string> DeserializeUrls(string json)
|
||||
{
|
||||
try { return JsonConvert.DeserializeObject<List<string>>(json) ?? new(); }
|
||||
catch { return new(); }
|
||||
}
|
||||
|
||||
private static async Task BackfillLegacyRedownloadsAsync(IServiceProvider services)
|
||||
{
|
||||
var db = services.GetRequiredService<ISqlSugarClient>();
|
||||
var legacy = await db.Queryable<DouyinReDownload>().Where(x => string.IsNullOrEmpty(x.TaskId)).ToListAsync();
|
||||
if (legacy.Count == 0) return;
|
||||
var tasks = services.GetRequiredService<VideoTaskService>();
|
||||
var task = await tasks.CreateTaskAsync(VideoTaskType.Redownload, VideoTaskTrigger.UpgradeRecovery,
|
||||
$"升级前重新下载（{legacy.Count} 条）");
|
||||
foreach (var job in legacy)
|
||||
{
|
||||
var video = !string.IsNullOrWhiteSpace(job.VideoRecordId)
|
||||
? await db.Queryable<DouyinVideo>().InSingleAsync(job.VideoRecordId)
|
||||
: await db.Queryable<DouyinVideo>().Where(x => x.AwemeId == job.ViedoId).FirstAsync();
|
||||
var item = await tasks.AddItemAsync(task.Id, job.ViedoId, job.CookieId, null, video?.ViedoType,
|
||||
video?.VideoTitle, video?.Author, job.SavePath, new[] { video?.VideoUrl }, video, video?.FileSize ?? 0);
|
||||
item.Stage = job.Status switch
|
||||
{
|
||||
1 => VideoTaskItemStage.Succeeded,
|
||||
2 => VideoTaskItemStage.Failed,
|
||||
_ => VideoTaskItemStage.Pending
|
||||
};
|
||||
item.ErrorMessage = job.ErrorMessage;
|
||||
item.ErrorType = job.Status == 2 ? VideoTaskErrorType.Unknown : VideoTaskErrorType.None;
|
||||
item.CompletedAt = job.Status == 0 ? null : job.UpdateTime;
|
||||
item.UpdatedAt = job.UpdateTime == default ? DateTime.Now : job.UpdateTime;
|
||||
await db.Updateable(item).ExecuteCommandAsync();
|
||||
job.TaskId = task.Id;
|
||||
job.TaskItemId = item.Id;
|
||||
}
|
||||
await db.Updateable(legacy).ExecuteCommandAsync();
|
||||
await tasks.RefreshCountsAsync(task.Id);
|
||||
var pending = legacy.Count(x => x.Status == 0);
|
||||
if (pending == 0) await tasks.CompleteTaskAsync(task.Id);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user