Реалізація двосторонньої синхронізації каталогу товарів з 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.