food-market/src/food-market.infrastructure/Integrations/MoySklad/MoySkladDocumentImportService.cs
nns 9506f9edc8 perf(migration): резюм документов по offset (не перечитывать уже перенесённое)
startOffset у всех документных импортов + эндпоинт ?offset=N: при продолжении
пропускаем уже перенесённые через offset API (а не перечитывая сотни страниц).
Идемпотентный skip по msId держит границу.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-09 15:52:28 +05:00

327 lines
24 KiB
C#
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

using System.Globalization;
using foodmarket.Application.Common.Tenancy;
using foodmarket.Application.Inventory;
using foodmarket.Domain.Inventory;
using foodmarket.Infrastructure.Persistence;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
namespace foodmarket.Infrastructure.Integrations.MoySklad;
/// <summary>Перенос ДОКУМЕНТОВ из МойСклада с историей (идемпотентно по id МойСклада,
/// резюмируемо — повторный запуск пропускает уже перенесённое). Даты moment переводятся
/// из МСК (UTC+3) в UTC; суммы из тиынов. Справочники (товары/контрагенты/группы) должны
/// быть импортированы ранее (несут MoySkladExternalId для сопоставления ссылок).
///
/// Реализован supply (приёмка) как ПРОВЕРЕННЫЙ ШАБЛОН; остальные типы (enter/loss/
/// retaildemand/…) добавляются по той же схеме.</summary>
public sealed class MoySkladDocumentImportService
{
private readonly MoySkladClient _client;
private readonly AppDbContext _db;
private readonly ITenantContext _tenant;
private readonly IStockService _stock;
private readonly ILogger<MoySkladDocumentImportService> _log;
public MoySkladDocumentImportService(MoySkladClient client, AppDbContext db, ITenantContext tenant,
IStockService stock, ILogger<MoySkladDocumentImportService> log)
{ _client = client; _db = db; _tenant = tenant; _stock = stock; _log = log; }
public sealed class DocImportProgress
{
public string Entity = "";
public int Seen, Created, Skipped, Errors;
public List<string> ErrorSamples { get; } = new();
}
/// <summary>МойСклад moment "yyyy-MM-dd HH:mm:ss[.fff]" (МСК) → UTC.</summary>
private static DateTime MomentToUtc(string? moment)
{
if (string.IsNullOrWhiteSpace(moment)) return DateTime.UtcNow;
var s = moment.Trim();
foreach (var fmt in new[] { "yyyy-MM-dd HH:mm:ss.fff", "yyyy-MM-dd HH:mm:ss" })
if (DateTime.TryParseExact(s, fmt, CultureInfo.InvariantCulture, DateTimeStyles.None, out var dt))
return DateTime.SpecifyKind(dt.AddHours(-3), DateTimeKind.Utc);
return DateTime.SpecifyKind(DateTime.Parse(s, CultureInfo.InvariantCulture).AddHours(-3), DateTimeKind.Utc);
}
private static string? IdFromHref(string? href)
{
if (string.IsNullOrEmpty(href)) return null;
var noQuery = href.Split('?')[0];
var seg = noQuery.TrimEnd('/').Split('/').LastOrDefault();
return string.IsNullOrEmpty(seg) ? null : seg;
}
// ── Приёмки (supply) ────────────────────────────────────────────────────
public async Task<DocImportProgress> ImportSuppliesAsync(string token, int startOffset,
DocImportProgress progress, CancellationToken ct)
{
var orgId = _tenant.OrganizationId ?? throw new InvalidOperationException("Tenant not set.");
progress.Entity = "supply";
var currencyId = await _db.Currencies.Select(c => c.Id).FirstAsync(ct);
var storeId = await EnsureStoreAsync(orgId, ct);
var adminUserId = await _db.Organizations.Where(o => o.Id == orgId).Select(o => o.AccountOwnerUserId).FirstOrDefaultAsync(ct);
var suppliers = await _db.Counterparties.Where(c => c.MoySkladExternalId != null)
.Select(c => new { c.Id, c.MoySkladExternalId }).ToDictionaryAsync(x => x.MoySkladExternalId!, x => x.Id, ct);
var fallbackSupplier = await _db.Counterparties.Select(c => (Guid?)c.Id).FirstOrDefaultAsync(ct);
var products = await _db.Products.Where(p => p.MoySkladExternalId != null)
.Select(p => new { p.Id, p.MoySkladExternalId }).ToDictionaryAsync(x => x.MoySkladExternalId!, x => x.Id, ct);
var done = (await _db.Supplies.Where(s => s.MoySkladExternalId != null)
.Select(s => s.MoySkladExternalId!).ToListAsync(ct)).ToHashSet();
// Номера приёмок МойСклада ПОВТОРЯЮТСЯ по годам, а у нас (OrganizationId, Number)
// уникален. Сохраняем оригинальный номер где можно, при коллизии добавляем год/счётчик.
var usedNumbers = (await _db.Supplies.Select(s => s.Number).ToListAsync(ct)).ToHashSet(StringComparer.OrdinalIgnoreCase);
string UniqueNumber(string baseName, int year)
{
if (usedNumbers.Add(baseName)) return baseName;
var withYear = $"{baseName}-{year}";
if (usedNumbers.Add(withYear)) return withYear;
for (var n = 2; ; n++) { var cand = $"{baseName}-{year}-{n}"; if (usedNumbers.Add(cand)) return cand; }
}
// expand=positions (НЕ positions.assortment — это раскрывает весь товар и делает
// страницу гигантской/зависающей). assortment.meta.href в позиции есть и так — из него
// берём id товара.
await foreach (var d in _client.StreamDocumentsAsync<MsDocument>(token, "entity/supply",
"positions,agent,store", startOffset, null, ct))
{
progress.Seen++;
if (d.Id is null || done.Contains(d.Id)) { progress.Skipped++; continue; }
try
{
var supplierMs = IdFromHref(d.Agent?.Meta?.Href);
var supplierId = supplierMs is not null && suppliers.TryGetValue(supplierMs, out var sid) ? sid
: fallbackSupplier ?? throw new InvalidOperationException("Нет контрагента-поставщика");
var momentUtc = MomentToUtc(d.Moment);
var posted = d.Applicable;
var baseNumber = string.IsNullOrWhiteSpace(d.Name) ? $"МС-{d.Id[..8]}" : d.Name!;
var supply = new foodmarket.Domain.Purchases.Supply
{
OrganizationId = orgId,
Number = UniqueNumber(baseNumber, momentUtc.Year),
Date = momentUtc,
Status = posted ? foodmarket.Domain.Purchases.SupplyStatus.Posted : foodmarket.Domain.Purchases.SupplyStatus.Draft,
SupplierId = supplierId,
StoreId = storeId,
CurrencyId = currencyId,
Notes = d.Description,
Total = d.Sum / 100m,
MoySkladExternalId = d.Id,
PostedAt = posted ? momentUtc : null,
PostedByUserId = posted ? adminUserId : null,
};
// Агрегируем позиции по товару (один и тот же товар в документе — суммируем
// кол-во, цена — средневзвешенная): иначе два движения по одному товару дадут
// дубликат строки stocks в рамках одного SaveChanges.
var byProduct = new Dictionary<Guid, (decimal Qty, decimal Amount)>();
foreach (var pos in d.Positions?.Rows ?? new())
{
var prodMs = IdFromHref(pos.Assortment?.Meta?.Href);
if (prodMs is null || !products.TryGetValue(prodMs, out var pid)) continue; // услуга/неизвестный
var up = pos.Price / 100m;
var cur = byProduct.GetValueOrDefault(pid);
byProduct[pid] = (cur.Qty + pos.Quantity, cur.Amount + pos.Quantity * up);
}
foreach (var (pid, agg) in byProduct)
{
var unit = agg.Qty != 0 ? decimal.Round(agg.Amount / agg.Qty, 4) : 0m;
supply.Lines.Add(new foodmarket.Domain.Purchases.SupplyLine
{
OrganizationId = orgId, ProductId = pid, Quantity = agg.Qty, UnitPrice = unit,
LineTotal = decimal.Round(agg.Amount, 2),
});
}
_db.Supplies.Add(supply);
if (posted)
foreach (var l in supply.Lines)
await _stock.ApplyMovementAsync(new StockMovementDraft(
ProductId: l.ProductId, StoreId: storeId, Quantity: l.Quantity,
Type: MovementType.Supply, DocumentType: "supply", DocumentId: supply.Id,
DocumentNumber: supply.Number, UnitCost: l.UnitPrice, OccurredAt: momentUtc,
CreatedByUserId: adminUserId), ct);
// Сохраняем ПО ОДНОМУ документу: следующий документ увидит уже
// закоммиченные строки stocks (ApplyMovementAsync ищет в БД) — без дублей.
await _db.SaveChangesAsync(ct);
_db.ChangeTracker.Clear(); // не копим tracked-граф на 7k документов
done.Add(d.Id);
progress.Created++;
}
catch (Exception ex)
{
// Критично: после сбоя SaveChanges очищаем трекер, иначе провалившиеся
// сущности остаются tracked и роняют ВСЕ последующие документы (каскад).
_db.ChangeTracker.Clear();
progress.Errors++;
if (progress.ErrorSamples.Count < 20) progress.ErrorSamples.Add($"{d.Name}/{d.Id}: {ex.Message}");
_log.LogWarning(ex, "supply import failed {Id}", d.Id);
}
}
return progress;
}
// ── общие помощники ──────────────────────────────────────────────────────
private static Dictionary<Guid, (decimal Qty, decimal Amount)> Aggregate(MsDocument d, Dictionary<string, Guid> products)
{
var byProduct = new Dictionary<Guid, (decimal Qty, decimal Amount)>();
foreach (var pos in d.Positions?.Rows ?? new())
{
var prodMs = IdFromHref(pos.Assortment?.Meta?.Href);
if (prodMs is null || !products.TryGetValue(prodMs, out var pid)) continue;
var up = pos.Price / 100m;
var cur = byProduct.GetValueOrDefault(pid);
byProduct[pid] = (cur.Qty + pos.Quantity, cur.Amount + pos.Quantity * up);
}
return byProduct;
}
private sealed record DocMaps(Guid StoreId, Guid CurrencyId, Guid? AdminUserId,
Dictionary<string, Guid> Products, Dictionary<string, Guid> Counterparties, Guid? FallbackCp);
private async Task<DocMaps> LoadMapsAsync(Guid orgId, CancellationToken ct)
{
var currencyId = await _db.Currencies.Select(c => c.Id).FirstAsync(ct);
var storeId = await EnsureStoreAsync(orgId, ct);
var admin = await _db.Organizations.Where(o => o.Id == orgId).Select(o => o.AccountOwnerUserId).FirstOrDefaultAsync(ct);
var cps = await _db.Counterparties.Where(c => c.MoySkladExternalId != null)
.Select(c => new { c.Id, c.MoySkladExternalId }).ToDictionaryAsync(x => x.MoySkladExternalId!, x => x.Id, ct);
var fb = await _db.Counterparties.Select(c => (Guid?)c.Id).FirstOrDefaultAsync(ct);
var prods = await _db.Products.Where(p => p.MoySkladExternalId != null)
.Select(p => new { p.Id, p.MoySkladExternalId }).ToDictionaryAsync(x => x.MoySkladExternalId!, x => x.Id, ct);
return new DocMaps(storeId, currencyId, admin, prods, cps, fb);
}
// ── Оприходования (enter) ────────────────────────────────────────────────
public async Task<DocImportProgress> ImportEntersAsync(string token, int startOffset, DocImportProgress progress, CancellationToken ct)
{
var orgId = _tenant.OrganizationId!.Value; progress.Entity = "enter";
var m = await LoadMapsAsync(orgId, ct);
var done = (await _db.Enters.Where(x => x.MoySkladExternalId != null).Select(x => x.MoySkladExternalId!).ToListAsync(ct)).ToHashSet();
var used = (await _db.Enters.Select(x => x.Number).ToListAsync(ct)).ToHashSet(StringComparer.OrdinalIgnoreCase);
string Uniq(string b, int y) { if (used.Add(b)) return b; var w = $"{b}-{y}"; if (used.Add(w)) return w; for (var n = 2; ; n++) { var c = $"{b}-{y}-{n}"; if (used.Add(c)) return c; } }
await foreach (var d in _client.StreamDocumentsAsync<MsDocument>(token, "entity/enter", "positions,store", startOffset, null, ct))
{
progress.Seen++; if (d.Id is null || done.Contains(d.Id)) { progress.Skipped++; continue; }
try
{
var mom = MomentToUtc(d.Moment); var posted = d.Applicable;
var e = new foodmarket.Domain.Purchases.Enter { OrganizationId = orgId, Number = Uniq(d.Name ?? $"МС-{d.Id[..8]}", mom.Year), Date = mom,
Status = posted ? foodmarket.Domain.Purchases.EnterStatus.Posted : foodmarket.Domain.Purchases.EnterStatus.Draft,
StoreId = m.StoreId, CurrencyId = m.CurrencyId, Notes = d.Description, Total = d.Sum / 100m, MoySkladExternalId = d.Id,
PostedAt = posted ? mom : null, PostedByUserId = posted ? m.AdminUserId : null };
foreach (var (pid, a) in Aggregate(d, m.Products))
e.Lines.Add(new foodmarket.Domain.Purchases.EnterLine { OrganizationId = orgId, ProductId = pid, Quantity = a.Qty, UnitCost = a.Qty != 0 ? decimal.Round(a.Amount / a.Qty, 4) : 0m, LineTotal = decimal.Round(a.Amount, 2) });
_db.Enters.Add(e);
if (posted) foreach (var l in e.Lines) await _stock.ApplyMovementAsync(new StockMovementDraft(l.ProductId, m.StoreId, l.Quantity, MovementType.Enter, "enter", e.Id, e.Number, l.UnitCost, mom, m.AdminUserId), ct);
await _db.SaveChangesAsync(ct); _db.ChangeTracker.Clear(); done.Add(d.Id); progress.Created++;
}
catch (Exception ex) { _db.ChangeTracker.Clear(); progress.Errors++; if (progress.ErrorSamples.Count < 20) progress.ErrorSamples.Add($"{d.Name}: {ex.Message}"); _log.LogWarning(ex, "enter {Id}", d.Id); }
}
return progress;
}
// ── Списания (loss) ──────────────────────────────────────────────────────
public async Task<DocImportProgress> ImportLossesAsync(string token, int startOffset, DocImportProgress progress, CancellationToken ct)
{
var orgId = _tenant.OrganizationId!.Value; progress.Entity = "loss";
var m = await LoadMapsAsync(orgId, ct);
var done = (await _db.Losses.Where(x => x.MoySkladExternalId != null).Select(x => x.MoySkladExternalId!).ToListAsync(ct)).ToHashSet();
var used = (await _db.Losses.Select(x => x.Number).ToListAsync(ct)).ToHashSet(StringComparer.OrdinalIgnoreCase);
string Uniq(string b, int y) { if (used.Add(b)) return b; var w = $"{b}-{y}"; if (used.Add(w)) return w; for (var n = 2; ; n++) { var c = $"{b}-{y}-{n}"; if (used.Add(c)) return c; } }
await foreach (var d in _client.StreamDocumentsAsync<MsDocument>(token, "entity/loss", "positions,store", startOffset, null, ct))
{
progress.Seen++; if (d.Id is null || done.Contains(d.Id)) { progress.Skipped++; continue; }
try
{
var mom = MomentToUtc(d.Moment); var posted = d.Applicable;
var e = new foodmarket.Domain.Inventory.Loss { OrganizationId = orgId, Number = Uniq(d.Name ?? $"МС-{d.Id[..8]}", mom.Year), Date = mom,
Status = posted ? foodmarket.Domain.Inventory.LossStatus.Posted : foodmarket.Domain.Inventory.LossStatus.Draft,
StoreId = m.StoreId, CurrencyId = m.CurrencyId, Reason = foodmarket.Domain.Inventory.LossReason.Other, Notes = d.Description,
Total = d.Sum / 100m, MoySkladExternalId = d.Id, PostedAt = posted ? mom : null, PostedByUserId = posted ? m.AdminUserId : null };
foreach (var (pid, a) in Aggregate(d, m.Products))
e.Lines.Add(new foodmarket.Domain.Inventory.LossLine { OrganizationId = orgId, ProductId = pid, Quantity = a.Qty, UnitCost = a.Qty != 0 ? decimal.Round(a.Amount / a.Qty, 4) : 0m, LineTotal = decimal.Round(a.Amount, 2) });
_db.Losses.Add(e);
if (posted) foreach (var l in e.Lines) await _stock.ApplyMovementAsync(new StockMovementDraft(l.ProductId, m.StoreId, -l.Quantity, MovementType.WriteOff, "loss", e.Id, e.Number, l.UnitCost, mom, m.AdminUserId), ct);
await _db.SaveChangesAsync(ct); _db.ChangeTracker.Clear(); done.Add(d.Id); progress.Created++;
}
catch (Exception ex) { _db.ChangeTracker.Clear(); progress.Errors++; if (progress.ErrorSamples.Count < 20) progress.ErrorSamples.Add($"{d.Name}: {ex.Message}"); _log.LogWarning(ex, "loss {Id}", d.Id); }
}
return progress;
}
// ── Возвраты поставщику (purchasereturn) ─────────────────────────────────
public async Task<DocImportProgress> ImportSupplierReturnsAsync(string token, int startOffset, DocImportProgress progress, CancellationToken ct)
{
var orgId = _tenant.OrganizationId!.Value; progress.Entity = "purchasereturn";
var m = await LoadMapsAsync(orgId, ct);
var done = (await _db.SupplierReturns.Where(x => x.MoySkladExternalId != null).Select(x => x.MoySkladExternalId!).ToListAsync(ct)).ToHashSet();
var used = (await _db.SupplierReturns.Select(x => x.Number).ToListAsync(ct)).ToHashSet(StringComparer.OrdinalIgnoreCase);
string Uniq(string b, int y) { if (used.Add(b)) return b; var w = $"{b}-{y}"; if (used.Add(w)) return w; for (var n = 2; ; n++) { var c = $"{b}-{y}-{n}"; if (used.Add(c)) return c; } }
await foreach (var d in _client.StreamDocumentsAsync<MsDocument>(token, "entity/purchasereturn", "positions,agent,store", startOffset, null, ct))
{
progress.Seen++; if (d.Id is null || done.Contains(d.Id)) { progress.Skipped++; continue; }
try
{
var mom = MomentToUtc(d.Moment); var posted = d.Applicable;
var cpMs = IdFromHref(d.Agent?.Meta?.Href);
var cpId = cpMs is not null && m.Counterparties.TryGetValue(cpMs, out var c1) ? c1 : m.FallbackCp!.Value;
var e = new foodmarket.Domain.Purchases.SupplierReturn { OrganizationId = orgId, Number = Uniq(d.Name ?? $"МС-{d.Id[..8]}", mom.Year), Date = mom,
Status = posted ? foodmarket.Domain.Purchases.SupplierReturnStatus.Posted : foodmarket.Domain.Purchases.SupplierReturnStatus.Draft,
SupplierId = cpId, StoreId = m.StoreId, CurrencyId = m.CurrencyId, Notes = d.Description, Total = d.Sum / 100m,
MoySkladExternalId = d.Id, PostedAt = posted ? mom : null, PostedByUserId = posted ? m.AdminUserId : null };
foreach (var (pid, a) in Aggregate(d, m.Products))
e.Lines.Add(new foodmarket.Domain.Purchases.SupplierReturnLine { OrganizationId = orgId, ProductId = pid, Quantity = a.Qty, UnitPrice = a.Qty != 0 ? decimal.Round(a.Amount / a.Qty, 4) : 0m, LineTotal = decimal.Round(a.Amount, 2) });
_db.SupplierReturns.Add(e);
if (posted) foreach (var l in e.Lines) await _stock.ApplyMovementAsync(new StockMovementDraft(l.ProductId, m.StoreId, -l.Quantity, MovementType.SupplierReturn, "supplier-return", e.Id, e.Number, l.UnitPrice, mom, m.AdminUserId), ct);
await _db.SaveChangesAsync(ct); _db.ChangeTracker.Clear(); done.Add(d.Id); progress.Created++;
}
catch (Exception ex) { _db.ChangeTracker.Clear(); progress.Errors++; if (progress.ErrorSamples.Count < 20) progress.ErrorSamples.Add($"{d.Name}: {ex.Message}"); _log.LogWarning(ex, "purchasereturn {Id}", d.Id); }
}
return progress;
}
// ── Розничные продажи (retaildemand) ─────────────────────────────────────
public async Task<DocImportProgress> ImportRetailSalesAsync(string token, int startOffset, DocImportProgress progress, CancellationToken ct)
{
var orgId = _tenant.OrganizationId!.Value; progress.Entity = "retaildemand";
var m = await LoadMapsAsync(orgId, ct);
var done = (await _db.RetailSales.Where(x => x.MoySkladExternalId != null).Select(x => x.MoySkladExternalId!).ToListAsync(ct)).ToHashSet();
var used = (await _db.RetailSales.Select(x => x.Number).ToListAsync(ct)).ToHashSet(StringComparer.OrdinalIgnoreCase);
string Uniq(string b, int y) { if (used.Add(b)) return b; var w = $"{b}-{y}"; if (used.Add(w)) return w; for (var n = 2; ; n++) { var c = $"{b}-{y}-{n}"; if (used.Add(c)) return c; } }
await foreach (var d in _client.StreamDocumentsAsync<MsDocument>(token, "entity/retaildemand", "positions,store", startOffset, null, ct))
{
progress.Seen++; if (d.Id is null || done.Contains(d.Id)) { progress.Skipped++; continue; }
try
{
var mom = MomentToUtc(d.Moment); var posted = d.Applicable;
var cash = d.CashSum / 100m; var card = d.NoCashSum / 100m; var total = d.Sum / 100m;
var pay = cash > 0 && card > 0 ? foodmarket.Domain.Sales.PaymentMethod.Mixed : (card > 0 ? foodmarket.Domain.Sales.PaymentMethod.Card : foodmarket.Domain.Sales.PaymentMethod.Cash);
var e = new foodmarket.Domain.Sales.RetailSale { OrganizationId = orgId, Number = Uniq(d.Name ?? $"МС-{d.Id[..8]}", mom.Year), Date = mom,
Status = posted ? foodmarket.Domain.Sales.RetailSaleStatus.Posted : foodmarket.Domain.Sales.RetailSaleStatus.Draft,
StoreId = m.StoreId, CurrencyId = m.CurrencyId, CashierUserId = m.AdminUserId, Payment = pay, PaidCash = cash, PaidCard = card,
Subtotal = total, Total = total, Notes = d.Description, MoySkladExternalId = d.Id, IsReturn = false,
PostedAt = posted ? mom : null, PostedByUserId = posted ? m.AdminUserId : null };
foreach (var (pid, a) in Aggregate(d, m.Products))
e.Lines.Add(new foodmarket.Domain.Sales.RetailSaleLine { OrganizationId = orgId, ProductId = pid, Quantity = a.Qty, UnitPrice = a.Qty != 0 ? decimal.Round(a.Amount / a.Qty, 4) : 0m, LineTotal = decimal.Round(a.Amount, 2) });
_db.RetailSales.Add(e);
if (posted) foreach (var l in e.Lines) await _stock.ApplyMovementAsync(new StockMovementDraft(l.ProductId, m.StoreId, -l.Quantity, MovementType.RetailSale, "retail-sale", e.Id, e.Number, l.UnitPrice, mom, m.AdminUserId), ct);
await _db.SaveChangesAsync(ct); _db.ChangeTracker.Clear(); done.Add(d.Id); progress.Created++;
}
catch (Exception ex) { _db.ChangeTracker.Clear(); progress.Errors++; if (progress.ErrorSamples.Count < 20) progress.ErrorSamples.Add($"{d.Name}: {ex.Message}"); _log.LogWarning(ex, "retaildemand {Id}", d.Id); }
}
return progress;
}
/// <summary>Гарантирует склад орг с MoySkladExternalId = склад МойСклада (магазин
/// с одним складом). Берём первый склад орг и помечаем его msId при первом вызове.</summary>
private async Task<Guid> EnsureStoreAsync(Guid orgId, CancellationToken ct)
{
var store = await _db.Stores.FirstOrDefaultAsync(s => s.OrganizationId == orgId, ct)
?? throw new InvalidOperationException("В организации нет склада");
return store.Id;
}
}