Реализация двусторонней синхронизации каталога товаров с ERP-системой

Реализация двусторонней синхронизации каталога товаров с ERP-системой

ERP (SAP, Oracle NetSuite, Microsoft Dynamics, Odoo) — система управления ресурсами предприятия. В отличие от 1С, ERP-системы имеют зрелые REST API или SOAP/OData-интерфейсы, но каждая со своей моделью данных и логикой версионирования.

Варианты интеграционных протоколов

Паттерн интеграции: Event-Driven vs Polling

Polling — сайт периодически запрашивает изменения в ERP. Проще реализовать, но создаёт задержку.

Webhooks / Change Data Capture — ERP уведомляет сайт о каждом изменении. Минимальная задержка, но требует поддержки со стороны ERP.

Гибридный подход (рекомендуется): webhooks для критичных данных (цены, остатки), polling раз в час для менее срочных (описания, характеристики).

Пример интеграции с Odoo

import xmlrpc.client

class OdooConnector:
    def __init__(self, url, db, username, password):
        self.url = url
        self.db  = db
        # Аутентификация
        common = xmlrpc.client.ServerProxy(f'{url}/xmlrpc/2/common')
        self.uid = common.authenticate(db, username, password, {})
        self.models = xmlrpc.client.ServerProxy(f'{url}/xmlrpc/2/object')

    def get_products(self, since: datetime = None):
        domain = [['active', '=', True]]
        if since:
            domain.append(['write_date', '>', since.isoformat()])

        return self.models.execute_kw(
            self.db, self.uid, self.password,
            'product.template', 'search_read',
            [domain],
            {'fields': ['id', 'name', 'default_code', 'list_price',
                        'qty_available', 'categ_id', 'description_sale']}
        )

Обработка изменений

class ERPSyncService:
    def sync_products(self):
        last_sync = SyncState.get_last_sync('erp_products')
        products  = self.erp.get_products(since=last_sync)

        updated = 0
        for erp_product in products:
            product, created = Product.objects.update_or_create(
                erp_id=erp_product['id'],
                defaults={
                    'name':     erp_product['name'],
                    'sku':      erp_product.get('default_code', ''),
                    'price':    erp_product['list_price'],
                    'stock':    erp_product['qty_available'],
                    'category': self.map_category(erp_product['categ_id']),
                }
            )
            updated += 1

        SyncState.update_last_sync('erp_products', datetime.now())
        return updated

Сроки

Интеграция с конкретной ERP через API с двусторонней синхронизацией: 12–20 рабочих дней в зависимости от качества документации и доступности тестовой среды ERP.