fix: misc. improvements for CORE API importer, per flaky data

handle some edge cases better; let config dictate whether some
warnings should be logged etc.

rattail_corepos/datasync/rattail.py

@@ -2,7 +2,7 @@
 ################################################################################
 #
 #  Rattail -- Retail Software Framework
-#  Copyright © 2010-2023 Lance Edgar
+#  Copyright © 2010-2024 Lance Edgar
 #
 #  This file is part of Rattail.
 #
@@ -26,8 +26,6 @@ DataSync for Rattail DB
 
 from sqlalchemy import orm
 
-from corepos.db.office_op import Session as CoreSession, model as corepos
-
 from rattail.datasync import DataSyncImportConsumer
 
 
@@ -143,7 +141,7 @@ class FromCOREAPIToRattail(DataSyncImportConsumer):
             if len(fields) == 2:
                 sku, vendorID = fields
                 vendor_item = self.api.get_vendor_item(sku, vendorID)
-                if vendor_item:
+                if vendor_item and vendor_item.get('upc'):
                     return self.api.get_product(vendor_item['upc'])
 
 
@@ -154,7 +152,8 @@ class FromCOREPOSToRattailBase(DataSyncImportConsumer):
     handler_spec = 'rattail_corepos.importing.corepos.db:FromCOREPOSToRattail'
 
     def begin_transaction(self):
-        self.corepos_session = CoreSession()
+        corepos = self.app.get_corepos_handler()
+        self.corepos_session = corepos.make_session_office_op()
 
     def rollback_transaction(self):
         self.corepos_session.rollback()
@@ -172,16 +171,18 @@ class FromCOREPOSToRattailProducts(FromCOREPOSToRattailBase):
     """
 
     def get_host_object(self, session, change):
+        corepos = self.app.get_corepos_handler()
+        op_model = corepos.get_model_office_op()
 
         if change.payload_type == 'Product':
             try:
-                return self.corepos_session.query(corepos.Product)\
+                return self.corepos_session.query(op_model.Product)\
-                                           .filter(corepos.Product.upc == change.payload_key)\
+                                           .filter(op_model.Product.upc == change.payload_key)\
                                            .one()
             except orm.exc.NoResultFound:
                 pass
 
         else:
             # try to fetch CORE POS object via typical method
-            Model = getattr(corepos, change.payload_type)
+            Model = getattr(op_model, change.payload_type)
             return self.corepos_session.get(Model, int(change.payload_key))

rattail_corepos/importing/corepos/api.py

@@ -489,15 +489,25 @@ class ProductImporter(FromCOREPOSAPI, corepos_importing.model.ProductImporter):
         self.vendor_items_by_upc = {}
 
         def cache(item, i):
-            self.vendor_items_by_upc.setdefault(item['upc'], []).append(item)
+            if item.get('upc'):
+                self.vendor_items_by_upc.setdefault(item['upc'], []).append(item)
 
         self.progress_loop(cache, self.api.get_vendor_items(),
                            message="Caching CORE Vendor Items")
 
-        self.maxval_unit_size = self.app.maxval(model.Product.unit_size)
-
     def get_host_objects(self):
-        return self.api.get_products()
+        products = OrderedDict()
+
+        def collect(product, i):
+            if product['upc'] in products:
+                log.warning("duplicate UPC encountered for '%s'; will discard: %s",
+                            product['upc'], product)
+            else:
+                products[product['upc']] = product
+
+        self.progress_loop(collect, self.api.get_products(),
+                           message="Fetching product info from CORE-POS")
+        return list(products.values())
 
     def identify_product(self, corepos_product):
         model = self.config.get_model()
@@ -537,6 +547,7 @@ class ProductImporter(FromCOREPOSAPI, corepos_importing.model.ProductImporter):
         return self.api.get_vendor_items(upc=api_product['upc'])
 
     def normalize_host_object(self, product):
+        model = self.model
         if 'upc' not in product:
             log.warning("CORE-POS product has no UPC: %s", product)
             return
@@ -594,7 +605,8 @@ class ProductImporter(FromCOREPOSAPI, corepos_importing.model.ProductImporter):
                 'uom_abbreviation': (size_info['uom_abbrev'] or '').strip() or None,
             })
 
-            if data['unit_size'] and data['unit_size'] >= self.maxval_unit_size:
+            maxval = self.app.maxval(model.Product.unit_size)
+            if data['unit_size'] and data['unit_size'] >= maxval:
                 log.warning("unit_size too large (%s) for product %s, will use null instead: %s",
                             data['unit_size'], data['upc'], product)
                 data['unit_size'] = None
@@ -693,6 +705,10 @@ class ProductCostImporter(FromCOREPOSAPI, corepos_importing.model.ProductCostImp
                                                         model.Product,
                                                         key='item_id')
 
+    def should_warn_for_missing_vendor_id(self):
+        return self.config.getbool('rattail.importing.corepos.vendor_items.warn_for_missing_vendor_id',
+                                   default=True)
+
     def get_host_objects(self):
 
         # first we will cache API products by upc
@@ -707,13 +723,15 @@ class ProductCostImporter(FromCOREPOSAPI, corepos_importing.model.ProductCostImp
 
         # next we cache API vendor items, also by upc
         vendor_items = {}
+        warn_for_missing_vendor_id = self.should_warn_for_missing_vendor_id()
 
         def cache(item, i):
             if not item['upc']:
                 log.warning("CORE vendor item has no upc: %s", item)
                 return
             if item['vendorID'] == '0':
-                log.warning("CORE vendor item has no vendorID: %s", item)
+                logger = log.warning if warn_for_missing_vendor_id else log.debug
+                logger("CORE vendor item has no vendorID: %s", item)
                 return
             vendor_items.setdefault(item['upc'], []).append(item)
 
@@ -766,7 +784,9 @@ class ProductCostImporter(FromCOREPOSAPI, corepos_importing.model.ProductCostImp
         return self.api.get_product(item['upc'])
 
     def get_product(self, item):
-        item_id = item['upc']
+        item_id = item.get('upc')
+        if not item_id:
+            return
 
         if hasattr(self, 'products_by_item_id'):
             return self.products_by_item_id.get(item_id)
@@ -873,7 +893,7 @@ class ProductCostImporter(FromCOREPOSAPI, corepos_importing.model.ProductCostImp
         # vendor items and then filter locally
         items = [item
                  for item in self.api.get_vendor_items()
-                 if item['upc'] == product['upc']]
+                 if item.get('upc') == product['upc']]
 
         vendor_id = product['default_vendor_id']
         self.sort_these_vendor_items(items, vendor_id)
@@ -973,6 +993,18 @@ class MemberImporter(FromCOREPOSAPI, corepos_importing.model.MemberImporter):
         self.people_by_card_number = self.cache_model(model.Person, query=query,
                                                       key=card_number)
 
+        self.membership_type_number_non_member = self.get_membership_type_number_non_member()
+
+    def get_membership_type_number_non_member(self):
+        if hasattr(self, 'membership_type_number_non_member'):
+            return self.membership_type_number_non_member
+
+        return self.config.getint('corepos.membership_type.non_member')
+
+    def should_warn_for_unknown_membership_type(self):
+        return self.config.getbool('rattail.importing.corepos.warn_for_unknown_membership_type',
+                                   default=True)
+
     def get_host_objects(self):
         return self.get_core_members()
 
@@ -1030,8 +1062,9 @@ class MemberImporter(FromCOREPOSAPI, corepos_importing.model.MemberImporter):
         # important to import the full member info from CORE, so that
         # we have it to sync back.  therefore can't afford to "skip"
         # any member records here
-        if (member['memberStatus'] not in self.member_status_codes
+        memstatus = (member['memberStatus'] or '').upper() or None
-            and member['memberStatus'] not in self.non_member_status_codes):
+        if (memstatus not in self.member_status_codes
+            and memstatus not in self.non_member_status_codes):
             log.warning("unexpected status '%s' for member %s: %s",
                         member['memberStatus'], card_number, member)
 
@@ -1056,9 +1089,17 @@ class MemberImporter(FromCOREPOSAPI, corepos_importing.model.MemberImporter):
         typeno = int(member['customerTypeID'] or 0)
         memtype = self.get_membership_type_by_number(typeno)
         if not memtype:
-            log.warning("unknown customerTypeID (membership_type_number) '%s' for: %s",
+            typeno = self.get_membership_type_number_non_member()
-                        member['customerTypeID'], member)
+            if typeno is not None:
-            typeno = None
+                memtype = self.get_membership_type_by_number(typeno)
+                if not memtype:
+                    raise ValueError("configured membership type for non-members is invalid!")
+
+            logger = log.warning if self.should_warn_for_unknown_membership_type() else log.debug
+            logger("unknown customerTypeID (membership_type_number) '%s' for: %s",
+                   member['customerTypeID'], member)
+            if typeno is not None:
+                log.debug("(will override with membership_type_number: %s)", typeno)
 
         data = {
             'number': card_number,