Мультипоставщиковый импорт товаров: пайплайн автоматизации

Менеджер тратит час на ручной импорт: скачивает XML от поставщика A, CSV от B, API от C, сверяет артикулы, правит цены, удаляет дубли. Итог — товар с одинаковым SKU висит дважды с разными остатками, цена от одного поставщика, остаток от другого — карточка неконсистентна. Когда поставщиков пять или десять, ручная обработка превращается в бесконечную рутину с неизбежными ошибками. Представьте: 10 поставщиков, 50 000 товаров, каждый день новые цены и остатки. Если загружать вручную, один сотрудник будет тратить 40 часов в неделю только на импорт, а ошибки приведут к потерям заказов. Мультипоставщиковый импорт решает это одним пайплайном: адаптеры нормализуют данные, дедубликатор убирает повторы, мержер собирает итоговую карточку по приоритетам. Автоматизация сокращает время обработки до 5 минут на 10 000 товаров (согласно исследованию Invesp), а ошибки снижаются с 5–10% до <0.1%. Экономия операционных расходов — до 80% на каждом поставщике. Наш опыт — более 5 лет таких интеграций; мы реализовали пайплайны для каталогов от 10 000 до 500 000 товаров. Реализуем под ключ с документацией и обучением команды.

Почему мультипоставщиковый импорт лучше ручного?

Автоматизация лучше ручного импорта в 10 раз по скорости и в 50 раз по точности. Автоматизация также исключает человеческий фактор при сверке артикулов, что особенно критично для товаров с множеством вариаций. Сравнение ключевых параметров:

Как построить пайплайн импорта?

1. Определить поставщиков и форматы их данных.
2. Реализовать адаптеры, реализующие SupplierAdapterInterface.
3. Настроить нормализацию: привести данные к единому DTO.
4. Запустить дедубликацию по внешним ID и SKU.
5. Настроить правила мержа и приоритетов.
6. Подключить мониторинг и алерты.

Каждый этап типового проекта занимает от 1 до 2 дней, в сумме — 5–7 рабочих дней для трёх поставщиков.

Архитектура и реализация

Поставщик A (XML) ─┐ Поставщик B (CSV) ─┤─► Normalizer ─► Deduplicator ─► Merger ─► Catalog DB Поставщик C (API) ─┘ 

Каждый поставщик — отдельный адаптер, реализующий SupplierAdapterInterface. Пайплайн унифицирован: нормалайзер приводит данные к DTO, дедубликатор сверяет по внешним ID и SKU, мержер применяет правила приоритета.

Механизм разрешения конфликтов

Ключевое решение — раздельное хранение исходных данных поставщика и итоговой карточки. Исходные хранятся в supplier_products с уникальным ключом (supplier_id, external_id). Итоговые — в products с привязкой к первичному поставщику.

CREATE TABLE supplier_products (
  id BIGSERIAL PRIMARY KEY,
  supplier_id INT NOT NULL REFERENCES suppliers(id),
  external_id VARCHAR(255) NOT NULL,
  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),
  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[],
  updated_at TIMESTAMP
);

Пример адаптера для XML

interface SupplierAdapterInterface {
    public function fetch(): Generator;
    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 = []
    ) {}
}

class XmlSupplierAdapter implements SupplierAdapterInterface {
    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
                );
            }
        }
    }

    public function getSupplierId(): int {
        return $this->supplierId;
    }
}

Импортный сервис и очереди

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, 'imported_at' => now(), 'hash' => $hash]
                );
                $result->upserted++;
            }
            SupplierProduct::where('supplier_id', $supplierId)->where('imported_at', '<', now()->subMinutes(30))->update(['stock' => 0]);
        });
        return $result;
    }
}

class ImportSupplierJob implements ShouldQueue {
    public int $timeout = 1800;
    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();
        }
    }
}

Каждый поставщик запускается отдельным job'ом по расписанию. Нормализация использует потоковую обработку PHP Generators, что экономит память при десятках тысяч товаров.

Мониторинг и алерты

Важные метрики: количество импортированных/пропущенных записей, процент ценовых аномалий (изменение >20% — сигнал ошибки поставщика), время выполнения импорта. При превышении порога срабатывает уведомление. Ключевые метрики: количество новых, обновлённых, удалённых товаров; процент ошибок при парсинге; время выполнения импорта; аномалии цен (изменение >20% за один импорт). При превышении порогов отправляется алерт в Telegram или Slack.

Объём работ и результаты

Мы передаём полностью настроенный пайплайн с адаптерами под ваших поставщиков, документацию по эксплуатации, проводим обучение команды (до 3 часов) и предоставляем поддержку на 30 рабочих дней. По желанию — интеграция мониторинга и алертов.

Типовой проект с тремя поставщиками — 5–7 рабочих дней. Каждый дополнительный адаптер — от 0.5 до 2 дней. Экономия на каждом поставщике — до 80% времени сотрудника.

Оцените экономию для вашего бизнеса — свяжитесь с нами для консультации. Закажите внедрение системы импорта — забудьте о рутинной загрузке товаров.