Files
DiunaBI/src/Backend/DiunaBI.Plugins.Morska/Processors/T3SingleSourceProcessor.cs

340 lines
12 KiB
C#
Raw Normal View History

using Microsoft.EntityFrameworkCore;
using DiunaBI.Core.Models;
using DiunaBI.Core.Database.Context;
using DiunaBI.Core.Services;
2025-06-02 18:53:25 +02:00
using Microsoft.Extensions.Logging;
using Google.Apis.Sheets.v4;
namespace DiunaBI.Plugins.Morska.Processors;
2025-06-02 16:54:33 +02:00
public class T3SingleSourceProcessor : MorskaBaseProcessor
{
2025-06-02 16:54:33 +02:00
public override string ProcessorType => "T3.SingleSource";
2025-06-02 18:53:25 +02:00
2025-06-02 16:54:33 +02:00
private readonly AppDbContext _db;
2025-06-02 18:53:25 +02:00
private readonly SpreadsheetsResource.ValuesResource _googleSheetValues;
private readonly ILogger<T3SingleSourceProcessor> _logger;
2025-06-07 13:51:27 +02:00
// Configuration properties loaded from layer records
private int Year { get; set; }
private int Month { get; set; }
private string? SourceLayerName { get; set; }
private string? Source { get; set; }
private Layer? SourceImportWorker { get; set; }
2025-06-02 16:54:33 +02:00
public T3SingleSourceProcessor(
2025-06-02 18:53:25 +02:00
AppDbContext db,
SpreadsheetsResource.ValuesResource googleSheetValues,
ILogger<T3SingleSourceProcessor> logger)
2025-06-02 16:54:33 +02:00
{
_db = db;
2025-06-02 18:53:25 +02:00
_googleSheetValues = googleSheetValues;
_logger = logger;
2025-06-02 16:54:33 +02:00
}
2025-06-07 13:51:27 +02:00
2025-06-02 16:54:33 +02:00
public override void Process(Layer processWorker)
{
2025-06-07 13:51:27 +02:00
try
{
_logger.LogInformation("{ProcessorType}: Starting processing for {ProcessWorkerName} ({ProcessWorkerId})",
ProcessorType, processWorker.Name, processWorker.Id);
// Load configuration from layer records
LoadConfiguration(processWorker);
// Validate required configuration
ValidateConfiguration();
// Perform the actual processing
PerformProcessing(processWorker);
_logger.LogInformation("{ProcessorType}: Successfully completed processing for {ProcessWorkerName}",
ProcessorType, processWorker.Name);
}
catch (Exception e)
{
_logger.LogError(e, "{ProcessorType}: Failed to process {ProcessWorkerName} ({ProcessWorkerId})",
ProcessorType, processWorker.Name, processWorker.Id);
throw;
}
}
private void LoadConfiguration(Layer processWorker)
{
if (processWorker.Records == null)
{
throw new InvalidOperationException("ProcessWorker has no records");
}
// Load year and month
var yearStr = GetRecordValue(processWorker.Records, "Year");
if (string.IsNullOrEmpty(yearStr) || !int.TryParse(yearStr, out var year))
{
throw new InvalidOperationException("Year record not found or invalid");
}
Year = year;
var monthStr = GetRecordValue(processWorker.Records, "Month");
if (string.IsNullOrEmpty(monthStr) || !int.TryParse(monthStr, out var month))
{
throw new InvalidOperationException("Month record not found or invalid");
}
Month = month;
// Load source layer name
SourceLayerName = GetRecordValue(processWorker.Records, "SourceLayer");
if (string.IsNullOrEmpty(SourceLayerName))
{
throw new InvalidOperationException("SourceLayer record not found");
}
// Load source name
Source = GetRecordValue(processWorker.Records, "Source");
if (string.IsNullOrEmpty(Source))
{
2025-06-07 13:51:27 +02:00
throw new InvalidOperationException("Source record not found");
}
2025-06-07 13:51:27 +02:00
_logger.LogDebug("{ProcessorType}: Configuration loaded - Year: {Year}, Month: {Month}, SourceLayer: {SourceLayer}, Source: {Source}",
ProcessorType, Year, Month, SourceLayerName, Source);
}
private void ValidateConfiguration()
{
var errors = new List<string>();
if (Year < 2000 || Year > 3000) errors.Add($"Invalid year: {Year}");
if (Month < 1 || Month > 12) errors.Add($"Invalid month: {Month}");
if (string.IsNullOrEmpty(SourceLayerName)) errors.Add("SourceLayer is required");
if (string.IsNullOrEmpty(Source)) errors.Add("Source is required");
// Find source import worker
SourceImportWorker = _db.Layers.SingleOrDefault(x => x.Name == SourceLayerName && !x.IsDeleted && !x.IsCancelled);
if (SourceImportWorker == null)
{
2025-06-07 13:51:27 +02:00
errors.Add($"SourceImportWorker layer '{SourceLayerName}' not found");
}
2025-06-07 13:51:27 +02:00
if (errors.Any())
{
2025-06-07 13:51:27 +02:00
throw new InvalidOperationException($"Configuration validation failed: {string.Join(", ", errors)}");
}
2025-06-07 13:51:27 +02:00
_logger.LogDebug("{ProcessorType}: Configuration validation passed", ProcessorType);
}
private void PerformProcessing(Layer processWorker)
{
_logger.LogDebug("{ProcessorType}: Processing data for Year: {Year}, Month: {Month}, Source: {Source}",
ProcessorType, Year, Month, Source);
// Get or create processed layer
var processedLayer = GetOrCreateProcessedLayer(processWorker);
// Get data sources
var dataSources = GetDataSources();
// Process records
var newRecords = ProcessRecords(dataSources);
// Save results
SaveProcessedLayer(processedLayer, newRecords);
_logger.LogInformation("{ProcessorType}: Successfully processed {RecordCount} records for layer {LayerName} ({LayerId})",
ProcessorType, newRecords.Count, processedLayer.Name, processedLayer.Id);
}
private Layer GetOrCreateProcessedLayer(Layer processWorker)
{
2025-06-02 16:54:33 +02:00
var processedLayer = _db.Layers
2025-06-07 13:51:27 +02:00
.Where(x => x.ParentId == processWorker.Id && !x.IsDeleted && !x.IsCancelled)
.OrderByDescending(x => x.CreatedAt)
.FirstOrDefault();
if (processedLayer == null)
{
processedLayer = new Layer
{
Id = Guid.NewGuid(),
Type = LayerType.Processed,
ParentId = processWorker.Id,
2025-06-07 13:51:27 +02:00
Number = _db.Layers.Count() + 1,
CreatedById = Guid.Parse("F392209E-123E-4651-A5A4-0B1D6CF9FF9D"),
ModifiedById = Guid.Parse("F392209E-123E-4651-A5A4-0B1D6CF9FF9D"),
CreatedAt = DateTime.UtcNow,
ModifiedAt = DateTime.UtcNow
};
2025-06-07 13:51:27 +02:00
processedLayer.Name = $"L{processedLayer.Number}-P-{Year}/{Month:D2}-{Source}-T3";
_logger.LogDebug("{ProcessorType}: Created new processed layer {LayerName}",
ProcessorType, processedLayer.Name);
}
else
{
processedLayer.ModifiedById = Guid.Parse("F392209E-123E-4651-A5A4-0B1D6CF9FF9D");
processedLayer.ModifiedAt = DateTime.UtcNow;
2025-06-07 13:51:27 +02:00
_logger.LogDebug("{ProcessorType}: Using existing processed layer {LayerName}",
ProcessorType, processedLayer.Name);
}
2025-06-07 13:51:27 +02:00
return processedLayer;
}
2025-06-07 13:51:27 +02:00
private List<Layer> GetDataSources()
{
2025-06-02 16:54:33 +02:00
var dataSources = _db.Layers
.Include(x => x.Records)
2025-06-07 13:51:27 +02:00
.Where(x => x.ParentId == SourceImportWorker!.Id && !x.IsDeleted && !x.IsCancelled)
.OrderBy(x => x.CreatedAt)
.AsNoTracking()
.ToList();
2025-06-07 13:51:27 +02:00
if (dataSources.Count == 0)
{
2025-06-07 13:51:27 +02:00
throw new InvalidOperationException($"No data sources found for import worker '{SourceImportWorker!.Name}'");
}
2025-06-07 13:51:27 +02:00
_logger.LogDebug("{ProcessorType}: Found {DataSourceCount} data sources for processing",
ProcessorType, dataSources.Count);
return dataSources;
}
private List<Record> ProcessRecords(List<Layer> dataSources)
{
var newRecords = new List<Record>();
var allRecords = dataSources.SelectMany(x => x.Records!).ToList();
2025-06-07 13:51:27 +02:00
_logger.LogDebug("{ProcessorType}: Processing records from {RecordCount} total records",
ProcessorType, allRecords.Count);
foreach (var baseRecord in dataSources.Last().Records!)
{
var codeRecords = allRecords.Where(x => x.Code == baseRecord.Code).ToList();
var processedRecord = new Record
{
Id = Guid.NewGuid(),
Code = baseRecord.Code,
CreatedAt = DateTime.UtcNow,
ModifiedAt = DateTime.UtcNow
};
2025-06-07 13:51:27 +02:00
ProcessDailyValues(processedRecord, codeRecords);
2025-06-07 13:51:27 +02:00
newRecords.Add(processedRecord);
}
_logger.LogDebug("{ProcessorType}: Processed {ProcessedRecordCount} records",
ProcessorType, newRecords.Count);
return newRecords;
}
private void ProcessDailyValues(Record processedRecord, List<Record> codeRecords)
{
var lastDayInMonth = DateTime.DaysInMonth(Year, Month);
// Day 1 - first value for the month
var firstVal = codeRecords
.Where(x => x.CreatedAt.Date <= new DateTime(Year, Month, 1))
.MaxBy(x => x.CreatedAt)?.Value1 ?? 0;
ProcessHelper.SetValue(processedRecord, 1, firstVal);
var previousValue = firstVal;
// Days 2 to last-1 - daily differences
for (var i = 2; i < lastDayInMonth; i++)
{
var dayVal = codeRecords
.Where(x => x.CreatedAt.Day == i && x.CreatedAt.Month == Month)
.MaxBy(x => x.CreatedAt)?.Value1;
if (dayVal == null)
{
2025-06-07 13:51:27 +02:00
ProcessHelper.SetValue(processedRecord, i, 0);
}
else
{
2025-06-07 13:51:27 +02:00
var processedVal = dayVal - previousValue;
ProcessHelper.SetValue(processedRecord, i, processedVal);
previousValue = (double)dayVal;
}
2025-06-07 13:51:27 +02:00
}
2025-06-07 13:51:27 +02:00
// Last day - special handling
var lastVal = codeRecords
.Where(x => x.CreatedAt.Date >= new DateTime(Year, Month, lastDayInMonth))
.MaxBy(x => x.CreatedAt)?.Value1;
2025-06-07 13:51:27 +02:00
if (lastVal == null)
{
ProcessHelper.SetValue(processedRecord, lastDayInMonth, 0);
}
else
{
ProcessHelper.SetValue(processedRecord, lastDayInMonth, (double)lastVal - previousValue);
}
2025-06-07 13:51:27 +02:00
// Copy last value to position 32
var valueToCopy = codeRecords.MaxBy(x => x.CreatedAt)?.Value1;
ProcessHelper.SetValue(processedRecord, 32, valueToCopy);
_logger.LogDebug("{ProcessorType}: Processed daily values for code {Code}, last value: {LastValue}",
ProcessorType, processedRecord.Code, valueToCopy);
}
private void SaveProcessedLayer(Layer processedLayer, List<Record> newRecords)
{
var isNew = processedLayer.Id == Guid.Empty || !_db.Layers.Any(x => x.Id == processedLayer.Id);
if (isNew)
{
2025-06-02 16:54:33 +02:00
_db.Layers.Add(processedLayer);
2025-06-07 13:51:27 +02:00
_logger.LogDebug("{ProcessorType}: Added new processed layer to database", ProcessorType);
}
else
{
2025-06-02 16:54:33 +02:00
_db.Layers.Update(processedLayer);
2025-06-07 13:51:27 +02:00
_logger.LogDebug("{ProcessorType}: Updated existing processed layer in database", ProcessorType);
}
2025-06-07 13:51:27 +02:00
2025-06-02 18:53:25 +02:00
SaveRecords(processedLayer.Id, newRecords);
2025-06-02 16:54:33 +02:00
_db.SaveChanges();
2025-06-07 13:51:27 +02:00
_logger.LogDebug("{ProcessorType}: Saved {RecordCount} records for layer {LayerId}",
ProcessorType, newRecords.Count, processedLayer.Id);
}
2025-06-07 13:51:27 +02:00
private void SaveRecords(Guid layerId, ICollection<Record> records)
2025-06-02 18:53:25 +02:00
{
2025-06-07 13:51:27 +02:00
// Remove existing records for this layer
2025-06-02 18:53:25 +02:00
var toDelete = _db.Records.Where(x => x.LayerId == layerId).ToList();
if (toDelete.Count > 0)
{
_db.Records.RemoveRange(toDelete);
2025-06-07 13:51:27 +02:00
_logger.LogDebug("{ProcessorType}: Removed {DeletedCount} existing records for layer {LayerId}",
ProcessorType, toDelete.Count, layerId);
2025-06-02 18:53:25 +02:00
}
2025-06-07 13:51:27 +02:00
// Add new records
2025-06-02 18:53:25 +02:00
foreach (var record in records)
{
record.CreatedById = Guid.Parse("F392209E-123E-4651-A5A4-0B1D6CF9FF9D");
record.CreatedAt = DateTime.UtcNow;
record.ModifiedById = Guid.Parse("F392209E-123E-4651-A5A4-0B1D6CF9FF9D");
record.ModifiedAt = DateTime.UtcNow;
record.LayerId = layerId;
_db.Records.Add(record);
}
2025-06-07 13:51:27 +02:00
_logger.LogDebug("{ProcessorType}: Added {RecordCount} new records for layer {LayerId}",
ProcessorType, records.Count, layerId);
2025-06-02 18:53:25 +02:00
}
2025-06-07 13:51:27 +02:00
private string? GetRecordValue(ICollection<Record> records, string code)
{
return records.FirstOrDefault(x => x.Code == code)?.Desc1;
}
}