Реализация мультипоставщикового импорта товаров на сайт
Когда каталог формируется из трёх и более поставщиков одновременно, простой «загрузить файл» перестаёт работать. Начинаются конфликты — одинаковые артикулы с разными ценами, дубли с разными SKU, поставщики с разными форматами данных. Мультипоставщиковый импорт требует унифицированного пайплайна с явными правилами разрешения конфликтов.
Архитектура пайплайна
Поставщик A (XML) ─┐
Поставщик B (CSV) ─┤─► Normalizer ─► Deduplicator ─► Merger ─► Catalog DB
Поставщик C (API) ─┘
Каждый поставщик — отдельный адаптер, выдающий на выходе унифицированный DTO. После нормализации данные проходят через единый пайплайн обработки.
Модель данных
Ключевое решение — хранить исходные данные поставщика отдельно от итоговой карточки товара.
-- Исходные данные от поставщиков (raw)
CREATE TABLE supplier_products (
id BIGSERIAL PRIMARY KEY,
supplier_id INT NOT NULL REFERENCES suppliers(id),
external_id VARCHAR(255) NOT NULL, -- ID в системе поставщика
sku VARCHAR(255),
barcode VARCHAR(50),
name TEXT NOT NULL,
price NUMERIC(12,2),
stock INT DEFAULT 0,
attributes JSONB DEFAULT '{}',
raw_data JSONB, -- исходный документ целиком
imported_at TIMESTAMP NOT NULL,
hash VARCHAR(64), -- SHA256 содержимого для детекции изменений
UNIQUE(supplier_id, external_id)
);
-- Итоговые карточки
CREATE TABLE products (
id BIGSERIAL PRIMARY KEY,
master_sku VARCHAR(255) UNIQUE,
name TEXT,
price NUMERIC(12,2),
stock INT,
primary_supplier_id INT REFERENCES suppliers(id),
merged_from INT[], -- supplier_product_id[]
updated_at TIMESTAMP
);
Адаптеры поставщиков
Каждый адаптер реализует единый интерфейс:
interface SupplierAdapterInterface
{
public function fetch(): Generator; // yields SupplierProductDTO
public function getSupplierId(): int;
}
class SupplierProductDTO
{
public function __construct(
public readonly string $externalId,
public readonly string $name,
public readonly float $price,
public readonly int $stock,
public readonly ?string $sku = null,
public readonly ?string $barcode = null,
public readonly array $attributes = [],
) {}
}
Адаптер для XML-поставщика:
class XmlSupplierAdapter implements SupplierAdapterInterface
{
public function __construct(
private readonly int $supplierId,
private readonly string $feedUrl,
) {}
public function fetch(): Generator
{
$reader = new XMLReader();
$reader->open($this->feedUrl);
while ($reader->read()) {
if ($reader->nodeType === XMLReader::ELEMENT && $reader->name === 'item') {
$node = new SimpleXMLElement($reader->readOuterXML());
yield new SupplierProductDTO(
externalId: (string) $node->id,
name: (string) $node->name,
price: (float) $node->price,
stock: (int) $node->quantity,
sku: (string) $node->article ?: null,
barcode: (string) $node->barcode ?: null,
);
}
}
}
public function getSupplierId(): int
{
return $this->supplierId;
}
}
Адаптер для CSV:
class CsvSupplierAdapter implements SupplierAdapterInterface
{
public function fetch(): Generator
{
$handle = fopen($this->filePath, 'r');
$headers = fgetcsv($handle, 0, ';');
$headers = array_map('trim', $headers);
while (($row = fgetcsv($handle, 0, ';')) !== false) {
$data = array_combine($headers, $row);
yield new SupplierProductDTO(
externalId: $data['ID'],
name: $data['Наименование'],
price: (float) str_replace(',', '.', $data['Цена']),
stock: (int) $data['Остаток'],
sku: $data['Артикул'] ?? null,
barcode: $data['Штрихкод'] ?? null,
);
}
fclose($handle);
}
}
Импортный сервис
class SupplierImportService
{
public function import(SupplierAdapterInterface $adapter): ImportResult
{
$supplierId = $adapter->getSupplierId();
$result = new ImportResult();
DB::transaction(function () use ($adapter, $supplierId, $result) {
foreach ($adapter->fetch() as $dto) {
$hash = hash('sha256', serialize($dto));
$existing = SupplierProduct::where([
'supplier_id' => $supplierId,
'external_id' => $dto->externalId,
])->first();
if ($existing && $existing->hash === $hash) {
$result->skipped++;
continue; // Данные не изменились
}
SupplierProduct::updateOrCreate(
['supplier_id' => $supplierId, 'external_id' => $dto->externalId],
[
'name' => $dto->name,
'price' => $dto->price,
'stock' => $dto->stock,
'sku' => $dto->sku,
'barcode' => $dto->barcode,
'attributes' => $dto->attributes,
'imported_at' => now(),
'hash' => $hash,
]
);
$result->upserted++;
}
// Отметить товары, исчезнувшие из последней выгрузки
SupplierProduct::where('supplier_id', $supplierId)
->where('imported_at', '<', now()->subMinutes(30))
->update(['stock' => 0]);
});
return $result;
}
}
Очереди и параллельный импорт
Каждый поставщик запускается как отдельный job:
class ImportSupplierJob implements ShouldQueue
{
public int $timeout = 1800; // 30 минут
public int $tries = 3;
public function handle(SupplierImportService $service): void
{
$adapter = SupplierAdapterFactory::make($this->supplier);
$result = $service->import($adapter);
Log::info("Supplier {$this->supplier->name} imported", $result->toArray());
// Запустить мерж после импорта всех активных поставщиков
if ($this->isLastActiveImport()) {
MergeProductsJob::dispatch();
}
}
}
Запуск по расписанию:
$schedule->job(new ImportSupplierJob($supplierA))->everyTwoHours();
$schedule->job(new ImportSupplierJob($supplierB))->everyTwoHours()->delay(5);
$schedule->job(new ImportSupplierJob($supplierC))->everyFourHours();
Мониторинг и алерты
Важные метрики для отслеживания:
- Количество импортированных / пропущенных / упавших записей
- Процент товаров, где цена изменилась на ±20% и более — вероятная ошибка поставщика
- Время выполнения импорта — рост означает увеличение каталога или деградацию источника
if ($result->priceAnomalies > $result->upserted * 0.05) {
Notification::send($admins, new SupplierPriceAnomalyAlert($supplier, $result));
}
Сроки реализации
- Базовая модель данных + 2 адаптера (XML + CSV): 2 дня
- Каждый дополнительный адаптер: +0.5–1 день
- Адаптер для REST API поставщика: +1–2 дня
- Очереди, логирование, алерты: +1 день
- Правила мержа и приоритизации: +1–2 дня (описано в отдельной услуге)
Типовой проект с тремя поставщиками и базовым мержем: 5–7 рабочих дней.