Implement Phase 2: end-to-end import system
Complete the import flow from file upload through duplicate resolution to transaction creation. Previously, ResolveDuplicatesAsync was a stub that only incremented counters, FileWatcherService detected files but didn't process them, and ImportsView.vue ignored the API response. Backend: - Add ParsedDataJson column to ImportBatch to persist parsed data between upload and confirm steps - Rewrite ImportService.ResolveDuplicatesAsync to deserialize parsed transactions and create/merge/skip based on user decisions, with payee matching, fingerprinting, and account balance updates - Add GetBatchesAsync for import history - Complete FileWatcherService: wait for file ready, resolve account by filename, call UploadAsync, send SignalR notification, archive file - Add GET /api/imports/batches endpoint Frontend: - Create imports API service (frontend/src/services/imports.ts) - Add ImportBatch, DuplicateResolution types - Rewrite ImportsView.vue with 3-step Vuetify stepper wizard: upload, review with duplicate handling, completion summary, plus import history table Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Catherine Renelle
@@ -1,16 +1,30 @@
|
||||
using Microsoft.AspNetCore.SignalR;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Purrse.Api.Hubs;
|
||||
using Purrse.Core.Interfaces.Services;
|
||||
using Purrse.Data;
|
||||
|
||||
namespace Purrse.Api.Services;
|
||||
|
||||
public class FileWatcherService : BackgroundService
|
||||
{
|
||||
private readonly IServiceScopeFactory _scopeFactory;
|
||||
private readonly ILogger<FileWatcherService> _logger;
|
||||
private readonly IHubContext<NotificationHub> _hubContext;
|
||||
private readonly string _importsPath;
|
||||
private FileSystemWatcher? _watcher;
|
||||
|
||||
public FileWatcherService(IServiceScopeFactory scopeFactory, ILogger<FileWatcherService> logger, IConfiguration config)
|
||||
private static readonly string[] SupportedExtensions = { ".ofx", ".qfx", ".csv", ".qif" };
|
||||
|
||||
public FileWatcherService(
|
||||
IServiceScopeFactory scopeFactory,
|
||||
ILogger<FileWatcherService> logger,
|
||||
IConfiguration config,
|
||||
IHubContext<NotificationHub> hubContext)
|
||||
{
|
||||
_scopeFactory = scopeFactory;
|
||||
_logger = logger;
|
||||
_hubContext = hubContext;
|
||||
_importsPath = config["Imports:WatchPath"] ?? Path.Combine(AppContext.BaseDirectory, "imports");
|
||||
}
|
||||
|
||||
@@ -19,10 +33,15 @@ public class FileWatcherService : BackgroundService
|
||||
if (!Directory.Exists(_importsPath))
|
||||
Directory.CreateDirectory(_importsPath);
|
||||
|
||||
var archivePath = Path.Combine(_importsPath, "archive");
|
||||
if (!Directory.Exists(archivePath))
|
||||
Directory.CreateDirectory(archivePath);
|
||||
|
||||
_watcher = new FileSystemWatcher(_importsPath)
|
||||
{
|
||||
NotifyFilter = NotifyFilters.FileName | NotifyFilters.CreationTime,
|
||||
EnableRaisingEvents = true
|
||||
EnableRaisingEvents = true,
|
||||
IncludeSubdirectories = false
|
||||
};
|
||||
|
||||
_watcher.Created += OnFileCreated;
|
||||
@@ -38,10 +57,109 @@ public class FileWatcherService : BackgroundService
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
private void OnFileCreated(object sender, FileSystemEventArgs e)
|
||||
private async void OnFileCreated(object sender, FileSystemEventArgs e)
|
||||
{
|
||||
_logger.LogInformation("New file detected: {FileName}", e.Name);
|
||||
// Files dropped here can be processed by the import service
|
||||
// A SignalR notification would be sent to connected clients
|
||||
try
|
||||
{
|
||||
var extension = Path.GetExtension(e.FullPath).ToLowerInvariant();
|
||||
if (!SupportedExtensions.Contains(extension))
|
||||
{
|
||||
_logger.LogDebug("Ignoring non-supported file: {FileName}", e.Name);
|
||||
return;
|
||||
}
|
||||
|
||||
_logger.LogInformation("New import file detected: {FileName}", e.Name);
|
||||
|
||||
// Wait for file to finish writing
|
||||
await WaitForFileReady(e.FullPath);
|
||||
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<PurrseDbContext>();
|
||||
var importService = scope.ServiceProvider.GetRequiredService<IImportService>();
|
||||
|
||||
// Find the target account by matching filename to account name
|
||||
var fileNameWithoutExt = Path.GetFileNameWithoutExtension(e.Name);
|
||||
var account = await db.Accounts
|
||||
.Where(a => a.IsActive && !a.IsClosed)
|
||||
.FirstOrDefaultAsync(a => a.Name.ToLower() == fileNameWithoutExt!.ToLower());
|
||||
|
||||
// Fall back to first active account
|
||||
account ??= await db.Accounts
|
||||
.Where(a => a.IsActive && !a.IsClosed)
|
||||
.OrderBy(a => a.SortOrder)
|
||||
.FirstOrDefaultAsync();
|
||||
|
||||
if (account == null)
|
||||
{
|
||||
_logger.LogWarning("No active account found for file import: {FileName}", e.Name);
|
||||
return;
|
||||
}
|
||||
|
||||
_logger.LogInformation("Importing {FileName} into account {AccountName} ({AccountId})",
|
||||
e.Name, account.Name, account.Id);
|
||||
|
||||
await using var fileStream = new FileStream(e.FullPath, FileMode.Open, FileAccess.Read, FileShare.Read);
|
||||
var result = await importService.UploadAsync(account.UserId, account.Id, e.Name!, fileStream);
|
||||
|
||||
_logger.LogInformation(
|
||||
"File import complete: {FileName} - {Total} transactions ({New} new, {Duplicates} duplicates)",
|
||||
e.Name, result.TotalTransactions, result.NewTransactions, result.PossibleDuplicates);
|
||||
|
||||
// Send SignalR notification to the account owner
|
||||
await _hubContext.Clients.Group(account.UserId.ToString()).SendAsync("FileImportReady", new
|
||||
{
|
||||
batchId = result.BatchId,
|
||||
fileName = e.Name,
|
||||
accountId = account.Id,
|
||||
accountName = account.Name,
|
||||
totalTransactions = result.TotalTransactions,
|
||||
newTransactions = result.NewTransactions,
|
||||
possibleDuplicates = result.PossibleDuplicates
|
||||
});
|
||||
|
||||
// Archive the processed file
|
||||
ArchiveFile(e.FullPath);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Error processing imported file: {FileName}", e.Name);
|
||||
}
|
||||
}
|
||||
|
||||
private static async Task WaitForFileReady(string filePath, int maxRetries = 30, int delayMs = 500)
|
||||
{
|
||||
for (int i = 0; i < maxRetries; i++)
|
||||
{
|
||||
try
|
||||
{
|
||||
using var stream = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.None);
|
||||
return; // File is ready
|
||||
}
|
||||
catch (IOException)
|
||||
{
|
||||
await Task.Delay(delayMs);
|
||||
}
|
||||
}
|
||||
|
||||
throw new IOException($"File {filePath} was not ready after {maxRetries * delayMs}ms");
|
||||
}
|
||||
|
||||
private void ArchiveFile(string filePath)
|
||||
{
|
||||
try
|
||||
{
|
||||
var archivePath = Path.Combine(_importsPath, "archive");
|
||||
var timestamp = DateTime.UtcNow.ToString("yyyyMMdd_HHmmss");
|
||||
var fileName = Path.GetFileName(filePath);
|
||||
var archiveFileName = $"{timestamp}_{fileName}";
|
||||
var destination = Path.Combine(archivePath, archiveFileName);
|
||||
|
||||
File.Move(filePath, destination);
|
||||
_logger.LogInformation("Archived import file to: {ArchivePath}", destination);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Failed to archive file: {FilePath}", filePath);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user