Інтеграція 1С з Magento 2

Інтеграція 1С з Magento 2

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

Архітектура інтеграції

1С ←→ Шар інтеграції ←→ Magento 2

Шар інтеграції:
- Черга (RabbitMQ / Redis)
- Трансформер (формат 1С ↔ формат Magento)
- Обробник помилок
- Retry логіка

Пряме з'єднання 1С → Magento REST API працює, але ненадійно при навантаженні. Проміжний шар з чергою — правильний підхід.

HTTP-сервіс 1С для експорту даних

// 1С: HTTP-сервіс ProductsService
// Метод GET /products
Function GetProducts(Request)
    MassiveOfGoods = New Array;

    Selection = Catalogs.Nomenclature.Select();
    While Selection.Next() Do
        If NOT Selection.IsFolder Then
            Good = New Structure;
            Good.Insert("sku",       String(Selection.Ref.UniqueId()));
            Good.Insert("name",      Selection.Name);
            Good.Insert("price",     InformationRegisters.PricesOfNomenclature.GetLast(Selection.Ref).Price);
            Good.Insert("quantity",  AccumulationRegisters.GoodsInWarehouses.BalanceOfGoods(Selection.Ref));
            MassiveOfGoods.Add(Good);
        EndIf;
    EndDo;

    Response = New HTTPServiceResponse(200);
    Response.SetBodyFromString(Format.JSON.Write(MassiveOfGoods));
    Return Response;
EndFunction

Імпорт з 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 тижні.