Интеграция 1С с Magento 2

Интеграция 1С с Magento 2

Двунаправленная синхронизация Magento 2 и 1С — стандартная задача для российского e-commerce: каталог и цены из 1С, заказы в 1С, остатки в реальном времени. Сложнее чем с OpenCart из-за Magento-специфичных понятий (attribute sets, configurable products, MSI).

Архитектура интеграции

1С ←→ Integration Layer ←→ Magento 2

Integration Layer:
- Queue (RabbitMQ / Redis)
- Transformer (1С-формат ↔ Magento-формат)
- Error Handler
- Retry Logic

Прямое соединение 1С → Magento REST API работает, но ненадёжно при нагрузке. Промежуточный слой с очередью — правильный подход.

1C HTTP-сервис для отдачи данных

// 1С: HTTP-сервис ProductsService
// Метод GET /products
Функция ПолучитьТовары(Запрос)
    МассивТоваров = Новый Массив;

    Выборка = Справочники.Номенклатура.Выбрать();
    Пока Выборка.Следующий() Цикл
        Если НЕ Выборка.ЭтоГруппа Тогда
            Товар = Новый Структура;
            Товар.Вставить("sku",            Строка(Выборка.Ссылка.УникальныйИдентификатор()));
            Товар.Вставить("name",           Выборка.Наименование);
            Товар.Вставить("price",          РегистрыСведений.ЦеныНоменклатуры.ПолучитьПоследнее(Выборка.Ссылка).Цена);
            Товар.Вставить("quantity",       РегистрыНакопления.ТоварыНаСкладах.ОстатокТовара(Выборка.Ссылка));
            МассивТоваров.Добавить(Товар);
        КонецЕсли;
    КонецЦикла;

    Ответ = Новый HTTPСервисОтвет(200);
    Ответ.УстановитьТелоИзСтроки(Формат.JSON.Записать(МассивТоваров));
    Возврат Ответ;
КонецФункции

Импорт из 1С в Magento 2 (Python integration worker)

import requests
import json
from magento.models.product import Product

class OneCMagentoSync:
    def __init__(self, m2_url, m2_token, onec_url, onec_auth):
        self.m2_url = m2_url
        self.m2_headers = {"Authorization": f"Bearer {m2_token}", "Content-Type": "application/json"}
        self.onec_url = onec_url
        self.onec_auth = onec_auth

    def sync_products(self):
        # Получить товары из 1С
        response = requests.get(f"{self.onec_url}/products", auth=self.onec_auth)
        products_1c = response.json()

        for product_data in products_1c:
            self.upsert_product(product_data)

    def upsert_product(self, data: dict):
        sku = data['sku']

        # Проверить существование в Magento
        check = requests.get(f"{self.m2_url}/rest/V1/products/{sku}", headers=self.m2_headers)

        payload = {
            "product": {
                "sku": sku,
                "name": data['name'],
                "price": float(data['price']),
                "status": 1,
                "type_id": "simple",
                "attribute_set_id": 4,
                "extension_attributes": {
                    "stock_item": {
                        "qty": int(data['quantity']),
                        "is_in_stock": int(data['quantity']) > 0
                    }
                }
            }
        }

        if check.status_code == 200:
            # Обновить
            requests.put(f"{self.m2_url}/rest/V1/products/{sku}",
                        headers=self.m2_headers, json=payload)
        else:
            # Создать
            requests.post(f"{self.m2_url}/rest/V1/products",
                         headers=self.m2_headers, json=payload)

Синхронизация остатков (MSI)

Magento 2.3+ использует Multi-Source Inventory. Обновление остатков через MSI API:

def update_stock_msi(self, sku: str, source_code: str, qty: float):
    payload = {
        "sourceItems": [{
            "sku": sku,
            "source_code": source_code,
            "quantity": qty,
            "status": 1 if qty > 0 else 0
        }]
    }
    requests.post(f"{self.m2_url}/rest/V1/inventory/source-items",
                 headers=self.m2_headers, json=payload)

Выгрузка заказов в 1С

def export_orders_to_1c(self):
    # Получить необработанные заказы
    response = requests.get(
        f"{self.m2_url}/rest/V1/orders",
        params={
            "searchCriteria[filter_groups][0][filters][0][field]": "status",
            "searchCriteria[filter_groups][0][filters][0][value]": "pending",
            "searchCriteria[pageSize]": 50
        },
        headers=self.m2_headers
    )

    orders = response.json()['items']

    for order in orders:
        order_1c = self.transform_order(order)

        # Отправить в 1С
        result = requests.post(
            f"{self.onec_url}/orders",
            json=order_1c,
            auth=self.onec_auth,
            timeout=10
        )

        if result.status_code == 200:
            # Отметить заказ как переданный в 1С
            requests.post(
                f"{self.m2_url}/rest/V1/orders/{order['entity_id']}/comments",
                headers=self.m2_headers,
                json={"statusHistory": {"comment": "Передан в 1С", "status": "processing"}}
            )

Очередь для надёжности

import pika  # RabbitMQ

def publish_to_queue(self, event_type: str, data: dict):
    connection = pika.BlockingConnection(pika.URLParameters(RABBITMQ_URL))
    channel = connection.channel()
    channel.queue_declare(queue='magento_1c_sync', durable=True)
    channel.basic_publish(
        exchange='',
        routing_key='magento_1c_sync',
        body=json.dumps({"type": event_type, "data": data}),
        properties=pika.BasicProperties(delivery_mode=2)  # persistent
    )
    connection.close()

Сроки

Базовая синхронизация (каталог + остатки в одну сторону) — 10–14 дней. Двусторонняя интеграция с заказами, очередью и мониторингом — 3–4 недели.