rattail/rattail-corepos
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

Add initial 'tax' support for Catapult export

Browse source
This commit is contained in:
Lance Edgar 2020-03-31 14:11:39 -05:00
parent e71e06f837
commit 2e5f7d3cd5
1 changed files with 42 additions and 31 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

63
rattail_corepos/corepos/importing/db/exporters/catapult_inventory.py
View file

@ -31,6 +31,7 @@ from sqlalchemy import orm
from corepos import enum as corepos_enum from corepos import enum as corepos_enum
from corepos.db.office_op import model as corepos from corepos.db.office_op import model as corepos
from corepos.db.util import table_exists
from rattail.gpc import GPC from rattail.gpc import GPC
from rattail.util import OrderedDict from rattail.util import OrderedDict
@ -139,15 +140,21 @@ class InventoryItemImporter(FromCore, catapult_importing.model.InventoryItemImpo
'corepos', 'exporting.catapult_inventory.warn_truncated_memo', 'corepos', 'exporting.catapult_inventory.warn_truncated_memo',
default=True) default=True)
try: self.floor_sections_exist = table_exists(self.host_session,
self.host_session.query(corepos.FloorSection).count() corepos.FloorSection)
except ProgrammingError as error: self.tax_components_exist = table_exists(self.host_session,
if "doesn't exist" in str(error): corepos.TaxRateComponent)
self.floor_sections_exist = False
else: self.tax_rate_ids_1 = self.config.getlist(
raise 'corepos', 'exporting.catapult_inventory.tax_rate_ids_1', default=[])
else: self.tax_rate_ids_1 = [int(id) for id in self.tax_rate_ids_1]
self.floor_sections_exist = True self.tax_rate_ids_2 = self.config.getlist(
'corepos', 'exporting.catapult_inventory.tax_rate_ids_2', default=[])
self.tax_rate_ids_2 = [int(id) for id in self.tax_rate_ids_2]
# TODO: should add component id levels too?
# tax_component_ids_1 = (1,)
# tax_component_ids_2 = (2,)
def query(self): def query(self):
query = self.host_session.query(corepos.Product)\ query = self.host_session.query(corepos.Product)\
@ -158,7 +165,8 @@ class InventoryItemImporter(FromCore, catapult_importing.model.InventoryItemImpo
.joinedload(corepos.VendorItem.vendor))\ .joinedload(corepos.VendorItem.vendor))\
.options(orm.joinedload(corepos.Product.default_vendor))\ .options(orm.joinedload(corepos.Product.default_vendor))\
.options(orm.joinedload(corepos.Product.scale_item))\ .options(orm.joinedload(corepos.Product.scale_item))\
.options(orm.joinedload(corepos.Product.user_info)) .options(orm.joinedload(corepos.Product.user_info))\
.options(orm.joinedload(corepos.Product.tax_rate))
if self.floor_sections_exist: if self.floor_sections_exist:
query = query.options(orm.joinedload(corepos.Product.physical_location)\ query = query.options(orm.joinedload(corepos.Product.physical_location)\
.joinedload(corepos.ProductPhysicalLocation.floor_section)) .joinedload(corepos.ProductPhysicalLocation.floor_section))
@ -181,6 +189,7 @@ class InventoryItemImporter(FromCore, catapult_importing.model.InventoryItemImpo
log.debug("ignoring UPC %s for product: %s", product.upc, product) log.debug("ignoring UPC %s for product: %s", product.upc, product)
return return
# convert item_id either to a PLU, or formatted UPC
is_plu = False is_plu = False
if len(str(int(item_id))) < 6: if len(str(int(item_id))) < 6:
is_plu = True is_plu = True
@ -192,8 +201,10 @@ class InventoryItemImporter(FromCore, catapult_importing.model.InventoryItemImpo
# drop leading zero(s) # drop leading zero(s)
if item_id[1] == '0': # UPC-A if item_id[1] == '0': # UPC-A
item_id = item_id[2:] item_id = item_id[2:]
assert len(item_id) == 12
else: # EAN13 else: # EAN13
item_id = item_id[1:] item_id = item_id[1:]
assert len(item_id) == 13
department = product.department department = product.department
if not department: if not department:
@ -218,31 +229,31 @@ class InventoryItemImporter(FromCore, catapult_importing.model.InventoryItemImpo
if product.scale and len(item_id) == 12 and item_id[0] == '2': if product.scale and len(item_id) == 12 and item_id[0] == '2':
weight_profile = 'LBNT' weight_profile = 'LBNT'
# TODO: need to finish the logic to map/calculate tax rates # calculate tax rates according to configured "mappings"
tax_1 = 0 tax_1 = 0
tax_2 = 0 tax_2 = 0
if product.tax_rate:
# # TODO: should let config drive these somehow, obviously # TODO: need to finish the logic to handle tax components
# tax_rate_ids_1 = (1,) if self.tax_components_exist and product.tax_rate.components:
# tax_rate_ids_2 = (2,)
# tax_component_ids_1 = (1,)
# tax_component_ids_2 = (2,)
# if product.tax_rate:
# if product.tax_rate.components:
# for component in product.tax_rate.components: # for component in product.tax_rate.components:
# if tax_component_ids_1 and component.id in tax_component_ids_1: # if tax_component_ids_1 and component.id in tax_component_ids_1:
# tax_1 += component.rate # tax_1 += component.rate
# if tax_component_ids_2 and component.id in tax_component_ids_2: # if tax_component_ids_2 and component.id in tax_component_ids_2:
# tax_2 += component.rate # tax_2 += component.rate
raise NotImplementedError
# else: # no components else: # no components
# rate = product.tax_rate rate = product.tax_rate
# if tax_rate_ids_1 and rate.id in tax_rate_ids_1: if self.tax_rate_ids_1 and rate.id in self.tax_rate_ids_1:
# tax_1 += rate.rate tax_1 += rate.rate
# if tax_rate_ids_2 and rate.id in tax_rate_ids_2: if self.tax_rate_ids_2 and rate.id in self.tax_rate_ids_2:
# tax_2 += rate.rate tax_2 += rate.rate
if not (self.tax_rate_ids_1 or self.tax_rate_ids_2) and rate.rate:
log.warning("product %s has unknown tax rate %s (%s) which will "
"be considered as tax 1: %s",
product.upc, rate.rate, rate.description, product)
tax_1 += rate.rate
location = None location = None
if self.floor_sections_exist and product.physical_location and product.physical_location.floor_section: if self.floor_sections_exist and product.physical_location and product.physical_location.floor_section: