467 lines
16 KiB
Python
467 lines
16 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
from pyjeeves.models.raw import (
|
|
Article as ArticleModel,
|
|
ProductClass, ArticleClass, CommodityGroup, ArticleEAN, ArticleUnit)
|
|
from pyjeeves.models import db
|
|
from sqlalchemy.sql.expression import and_
|
|
from sqlalchemy.orm.exc import NoResultFound
|
|
from gtin import GTIN
|
|
|
|
from pyjeeves import logging
|
|
logger = logging.getLogger("PyJeeves." + __name__)
|
|
|
|
|
|
# Relocate Jeeves modules to separate folder and let a "master" module handle imports, and setup.
|
|
class Article():
|
|
"""Handles articles in Jeeves, currently filters out all articles with class = 2"""
|
|
|
|
@staticmethod
|
|
def get(art_no):
|
|
""" Query an article by number """
|
|
try:
|
|
return db.raw_session.query(ArticleModel).filter_by(
|
|
ArtNr=str(art_no)
|
|
).one()
|
|
except NoResultFound:
|
|
raise KeyError
|
|
|
|
@staticmethod
|
|
def get_all(filter_=and_(
|
|
ArticleModel.ItemStatusCode == 0,
|
|
ArticleModel.ArtKod != 2,
|
|
ArticleModel.VaruGruppKod != 90,
|
|
ArticleModel.ArtProdKlass != 0)
|
|
):
|
|
# .filter_by(ItemStatusCode=0, ArtKod=2)
|
|
return db.raw_session.query(ArticleModel).filter(filter_).all()
|
|
|
|
@staticmethod
|
|
def get_article_units(filter_=and_()):
|
|
return db.raw_session.query(ArticleUnit).filter(filter_).all()
|
|
|
|
@staticmethod
|
|
def is_salable(art_no_list=[]):
|
|
""" Returns true if all articles are salable,
|
|
else false with error information """
|
|
articles = db.raw_session.query(ArticleModel).filter(
|
|
and_(ArticleModel.ArtNr.in_(art_no_list))).all()
|
|
|
|
blocked_articles = [article.ArtNr for article in articles
|
|
if article.ArtKod == 2 or article.ItemStatusCode != 0]
|
|
unknown_articles = [x for x in art_no_list
|
|
if x not in set([article.ArtNr for article in articles])]
|
|
|
|
if blocked_articles or unknown_articles:
|
|
errors = {}
|
|
if blocked_articles:
|
|
errors['blocked_articles'] = blocked_articles
|
|
if unknown_articles:
|
|
errors['unknown_articles'] = unknown_articles
|
|
return False, errors
|
|
|
|
return True, {}
|
|
|
|
@staticmethod
|
|
def get_article_gtins():
|
|
return db.raw_session.query(ArticleEAN).all()
|
|
|
|
@staticmethod
|
|
def add_article_gtins(gtins=[], dry_run=False):
|
|
# Expects a list of dicts like this:
|
|
# [{
|
|
# 'article_no': article.ArtNr,
|
|
# 'article_gtin': gtin,
|
|
# 'unit': unit.AltEnhetKod,
|
|
# }]
|
|
for gtin in gtins:
|
|
n1 = ArticleEAN(
|
|
ArtNr=gtin['article_no'], AltEnhetKod=gtin.get('unit', None),
|
|
ArtNrEAN=str(gtin['article_gtin']), ForetagKod=1)
|
|
if dry_run:
|
|
logger.info('Creating GTIN for %s, %s, %s' % (n1.ArtNr, n1.AltEnhetKod, n1.ArtNrEAN))
|
|
continue
|
|
|
|
db.raw_db.add(n1)
|
|
logger.debug('Created/updated Article EAN for %s - %s with GTIN %s' % (
|
|
gtin['article_no'], gtin.get('unit', 'no unit'), gtin['article_gtin']))
|
|
|
|
db.raw_db.commit()
|
|
logger.info('Succesfully commited %s GTINs to database' % (len(gtins)))
|
|
|
|
@staticmethod
|
|
def clear_article_gtins():
|
|
gtins = db.raw_session.query(ArticleEAN).all()
|
|
|
|
for gtin in gtins:
|
|
db.raw_db.delete(gtin)
|
|
|
|
db.raw_db.commit()
|
|
logger.info('Deleted %s GTINs' % (len(gtins)))
|
|
|
|
|
|
class ArticleCategory():
|
|
"""Handles article categories, such as classes and groups in Jeeves"""
|
|
|
|
@staticmethod
|
|
def get_all():
|
|
# .filter_by(ItemStatusCode=0, ArtKod=2)
|
|
prod_classes = db.raw_session.query(ProductClass).all()
|
|
art_classes = db.raw_session.query(ArticleClass).all()
|
|
com_groups = db.raw_session.query(CommodityGroup).all()
|
|
|
|
return {'ProductClasses': prod_classes,
|
|
'ArticleClasses': art_classes, 'CommodityGroups': com_groups}
|
|
|
|
|
|
# TODO: Should be moved to separate project with Lindvalls specific code
|
|
def get_gtin_for_article(article_ean, article_unit=None, use_prefix=True):
|
|
# If we don't want to prefix with 0, then exclude them here.
|
|
UNIT_MAPPING = {
|
|
'Påse': '',
|
|
'st': '',
|
|
'paket': 0,
|
|
'200g': 0,
|
|
'kg': 9,
|
|
'Kart': 1,
|
|
'Bricka': 1,
|
|
'½-pall': 2,
|
|
'tray_no_wrap': 8
|
|
}
|
|
prefixes = []
|
|
if article_unit:
|
|
# Find matching values in unit mapping
|
|
prefixes = [
|
|
val for key, val in UNIT_MAPPING.items()
|
|
if article_unit[0:len(key)].lower() in key.lower()]
|
|
if len(prefixes) > 1:
|
|
logger.warning('More than one unit match found in unit mapping')
|
|
|
|
# Use the first match
|
|
raw_gtin = (str(prefixes[0]) + article_ean) if prefixes and use_prefix else article_ean
|
|
|
|
# Handle GS1-128 GTIN code
|
|
if len(raw_gtin) >= 15 and raw_gtin[0:2] == '01':
|
|
raw_gtin = raw_gtin[2::]
|
|
|
|
article_gtin = GTIN(raw=raw_gtin)
|
|
|
|
return article_gtin
|
|
|
|
|
|
# TODO: Should be moved to separate project with Lindvalls specific code
|
|
def create_gtins_for_trading_goods(filename='gtin_trading_goods.csv'):
|
|
articles = Article.get_all(and_(
|
|
ArticleModel.ArtProdKlass == 4))
|
|
|
|
gtins = []
|
|
gtin_data = {}
|
|
|
|
import csv
|
|
with open(filename, newline='') as csvfile:
|
|
gtinreader = csv.reader(csvfile, delimiter=',')
|
|
headers = gtinreader.__next__()
|
|
logger.info('Found these columns: %s' % (', '.join(headers)))
|
|
|
|
for row in gtinreader:
|
|
gtin_data[row[0]] = row
|
|
|
|
logger.info("Found %s articles and updating with %s rows of data" % (
|
|
len(articles), len(gtin_data)))
|
|
|
|
for article in articles:
|
|
data = gtin_data.get(article.ArtNr)
|
|
if data:
|
|
default_set = False
|
|
if len(article.ArticleUnit) == 0 and data[3]:
|
|
logger.warning('Article %s has no ArticleUnits, but requires it' % (article.ArtNr))
|
|
|
|
for unit in article.ArticleUnit:
|
|
if unit.AltEnhetKod[0:3] == 'kart' and not data[3]:
|
|
logger.warning('Article %s missing kart unit' % (article.ArtNr))
|
|
|
|
if data[3] and unit.AltEnhetKod != 'st':
|
|
gtin = get_gtin_for_article(data[3], unit.AltEnhetKod, False)
|
|
# Only add GTINs for order units (not PO units)
|
|
if unit.AltEnhetOrder == '1':
|
|
gtins.append({
|
|
'article_no': article.ArtNr,
|
|
'article_gtin': gtin,
|
|
'unit': unit.AltEnhetKod
|
|
})
|
|
if unit.AltEnhetKod == 'st':
|
|
gtin = get_gtin_for_article(data[2], unit.AltEnhetKod, False)
|
|
# Only add GTINs for order units (not PO units)
|
|
if unit.AltEnhetOrder == '1':
|
|
gtins.append({
|
|
'article_no': article.ArtNr,
|
|
'article_gtin': gtin,
|
|
'unit': unit.AltEnhetKod
|
|
})
|
|
default_set = True
|
|
|
|
# Add default gtin if 'st' not used
|
|
if not default_set:
|
|
gtin = get_gtin_for_article(data[2], None, False)
|
|
gtins.append({
|
|
'article_no': article.ArtNr,
|
|
'article_gtin': gtin
|
|
})
|
|
else:
|
|
# Warn about active and stock items that didn't get updated.
|
|
if article.LagTyp == 0 and article.ItemStatusCode == 0:
|
|
logger.warning('Article %s has no GTIN data in CSV' % (article.ArtNr))
|
|
|
|
Article.add_article_gtins(gtins)
|
|
|
|
|
|
# TODO: Should be moved to separate project with Lindvalls specific code
|
|
def create_gtins(dry_run=True):
|
|
# GS1 Company Prefixes that we manage locally, prefixing etc.
|
|
LOCAL_GCPS = [
|
|
'731083', # Lindvalls Kaffe
|
|
'7392736', # Sackeus AB
|
|
'735007318', # Sarria Import AB
|
|
'732157', # Martin & Servera AB
|
|
'350096', # Scænsei Thee Kompani AB (Used by REKYL In Omnia Paratus AB)
|
|
'735003307', # Coffee Please
|
|
'735003711', # Emmas Skafferi AB (Used by Coffee Please)
|
|
'735003712', # Prefix no longer subscribed (Used by Coffee Please)
|
|
'735003302', # Josephine Selander - YogaGo (Used by Coffee Please)
|
|
]
|
|
|
|
articles = Article.get_all(and_(
|
|
ArticleModel.ItemStatusCode == 0,
|
|
ArticleModel.VaruGruppKod != 90,
|
|
ArticleModel.ArtProdKlass != 0))
|
|
|
|
articles_with_existing_gtins = [
|
|
gtin.ArtNr for i, gtin in enumerate(Article.get_article_gtins())]
|
|
|
|
gtins = []
|
|
|
|
for article in articles:
|
|
if article.ArtNr in articles_with_existing_gtins:
|
|
continue
|
|
|
|
if not article.ArtStreckKod:
|
|
logger.warning('No base GTIN for article %s' % (article.ArtNr))
|
|
continue
|
|
|
|
GCP = GTIN(raw=article.ArtStreckKod).gcp
|
|
|
|
if 12 < len(article.ArtStreckKod) < 12 and GCP in LOCAL_GCPS:
|
|
logger.error('Base GTIN is wrong length for article %s' % (article.ArtNr))
|
|
continue
|
|
|
|
# If GTIN is provided by vendor, skip prefixes and gohead if only one or no units exist.
|
|
if GCP not in LOCAL_GCPS and len(article.ArticleUnit) <= 1:
|
|
use_prefix = False
|
|
logger.info('Externally provided GTIN for %s, skipping prefixes' % (article.ArtNr))
|
|
elif GCP not in LOCAL_GCPS and len(article.ArticleUnit) > 1:
|
|
logger.warning('Externally provided GTIN for %s, too many units' % (article.ArtNr))
|
|
continue
|
|
else:
|
|
use_prefix = True
|
|
|
|
# Create gtin without ArticleUnit, for the base unit.
|
|
# gtins.append({
|
|
# 'article_no': article.ArtNr,
|
|
# 'article_gtin': get_gtin_for_article(article.ArtStreckKod, None, False)
|
|
# })
|
|
|
|
for unit in article.ArticleUnit:
|
|
|
|
# Skip paket for 21%, should only match HV with plastic wrapping.
|
|
if article.ArtNr[0:2] == '21' and unit.AltEnhetKod[0:6].lower() == 'paket':
|
|
logger.info('Skip paket unit for %s' % (article.ArtNr))
|
|
continue
|
|
|
|
# Special for 20%/30%, should only match HV without plastic wrapping.
|
|
if article.ArtNr[0:2] in ('20', '30') and unit.AltEnhetKod[0:6].lower() == 'bricka':
|
|
unit_code = 'tray_no_wrap'
|
|
else:
|
|
unit_code = unit.AltEnhetKod
|
|
|
|
gtin = get_gtin_for_article(article.ArtStreckKod, unit_code, use_prefix)
|
|
|
|
# Only add GTINs for order units (not PO units)
|
|
if unit.AltEnhetOrder == '1':
|
|
gtins.append({
|
|
'article_no': article.ArtNr,
|
|
'article_gtin': gtin,
|
|
'unit': unit.AltEnhetKod
|
|
})
|
|
|
|
# Workaround for scanning HV base units without plastic wrapping
|
|
if str(gtin)[0] == '0':
|
|
# Create gtin without ArticleUnit, for the base unit.
|
|
gtins.append({
|
|
'article_no': article.ArtNr,
|
|
'article_gtin': get_gtin_for_article(article.ArtStreckKod, None, False)
|
|
})
|
|
|
|
# Add GTIN to articles that don't use article units
|
|
# Should this still be added to arean/ArticleEAN???
|
|
# if len(article.ArticleUnit) == 0:
|
|
# gtin = get_gtin_for_article(article.ArtStreckKod, None, use_prefix)
|
|
# gtins.append({
|
|
# 'article_no': article.ArtNr,
|
|
# 'article_gtin': gtin,
|
|
# 'unit': None,
|
|
# })
|
|
# add_gtin_for_article(
|
|
# article.ArtNr, article.ArtStreckKod, None, use_prefix)
|
|
|
|
Article.add_article_gtins(gtins, dry_run)
|
|
|
|
|
|
# TODO: Should be moved to separate project with Lindvalls specific code
|
|
def find_articles_without_base_gtin():
|
|
articles = Article.get_all(and_(
|
|
ArticleModel.ItemStatusCode == 0,
|
|
ArticleModel.VaruGruppKod != 90,
|
|
ArticleModel.ArtProdKlass != 0))
|
|
|
|
_list = []
|
|
|
|
for article in articles:
|
|
if not article.ArtStreckKod:
|
|
_list.append(
|
|
{'artnr': article.ArtNr,
|
|
'artbeskr': article.ArtBeskr,
|
|
'error': 'no_base'})
|
|
continue
|
|
else:
|
|
if 12 < len(article.ArtStreckKod) < 12:
|
|
_list.append(
|
|
{'artnr': article.ArtNr,
|
|
'artbeskr': article.ArtBeskr,
|
|
'error': 'wrong_length'})
|
|
continue
|
|
|
|
for item in _list:
|
|
print('{artnr}, "{artbeskr}", {error}'.format(**item))
|
|
|
|
|
|
# TODO: Should be moved to separate project with Lindvalls specific code
|
|
def set_storage_type():
|
|
articles = Article.get_all(and_(
|
|
ArticleModel.LagTyp == 0,
|
|
ArticleModel.ItemStatusCode == 0,
|
|
ArticleModel.AnskaffningsSatt == 10))
|
|
|
|
for article in articles:
|
|
article.LagTyp = 4
|
|
|
|
db.raw_db.commit()
|
|
|
|
logger.info("Updated storage type for %s articles" % (len(articles)))
|
|
|
|
|
|
# TODO: Should be moved to separate project with Lindvalls specific code
|
|
def set_zone_placement():
|
|
# Logic for article groups and zones
|
|
# ArticleClass descides which zone to put it.
|
|
# set Article.ArticleBalance[0].japp_ewms_rec_zoneid to correct zoneid
|
|
|
|
article_class_map = {
|
|
'Kaffe': 'U',
|
|
'OoH-Kaffe': 'K',
|
|
'Private Label': 'S',
|
|
'Tillbehör och maskiner': 'U',
|
|
'Komplement': 'U'
|
|
}
|
|
|
|
articles = Article.get_all(and_(
|
|
ArticleModel.ItemStatusCode == 0,
|
|
ArticleModel.AnskaffningsSatt == 10))
|
|
|
|
zone_placements_update = 0
|
|
for article in articles:
|
|
zone_id = article_class_map.get(article.ArticleClass.ArtTypBeskr)
|
|
if zone_id and article.ArticleBalance:
|
|
article.ArticleBalance[0].JAPP_EWMS_REC_ZoneID = zone_id
|
|
zone_placements_update += 1
|
|
else:
|
|
logger.info("Excluded %s, wrong article class or no balance " % (article.ArtNr))
|
|
|
|
db.raw_db.commit()
|
|
|
|
logger.info("Updated placement zone for %s articles" % (zone_placements_update))
|
|
|
|
# a = Article.get('2109')
|
|
# print([ab.to_dict() for ab in a['ArticleBalance']])
|
|
|
|
|
|
def update_decimals_on_alt_units():
|
|
units = Article.get_article_units(ArticleUnit.AltEnhetKod == 'påse')
|
|
|
|
updated_units = 0
|
|
|
|
for unit in units:
|
|
if unit.AltEnhetOmrFaktor is not None:
|
|
dec_count = 0
|
|
for digit in unit.AltEnhetOmrFaktor.as_tuple().digits:
|
|
if digit != 0:
|
|
dec_count += 1
|
|
unit.AltEnhetAntDec = dec_count
|
|
|
|
updated_units += 1
|
|
|
|
db.raw_db.commit()
|
|
logger.info("Updated decimal count for %s article units" % (updated_units))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
# print([column.key for column in Company.__table__.columns])
|
|
|
|
# from pprint import pprint
|
|
# logger.info("Starting TEST")
|
|
|
|
# session = RawSession()
|
|
|
|
# logger.info("Testing gettings an article")
|
|
# # c1 = session.query(Company).filter_by(FtgNr="179580").first()
|
|
# # print(ArticleModel)
|
|
# c1 = db.raw_session.query(ArticleModel).filter_by(ArtNr="2003").first()
|
|
# c1 = Article.get("2003")
|
|
# pprint([unit.to_dict() for unit in c1.ArticleUnit])
|
|
# pprint(c1.to_dict())
|
|
# pprint([(au.to_dict(), au.AltEnhetOrder) for au in c1.ArticleUnit])
|
|
# logger.info(c1.to_dict())
|
|
|
|
# print(
|
|
# len(Article.get_all())
|
|
# )
|
|
|
|
# c1 = db.raw_session.query(ArticleEAN).all()
|
|
# pprint([c.to_dict() for c in c1])
|
|
|
|
# c1 = db.raw_session.query(ArticleEAN).filter_by(ArtNr="1054").first()
|
|
# pprint(c1.to_dict())
|
|
# c1.ArtNrEAN = '7310830010548'
|
|
# pprint(c1.to_dict())
|
|
# c1.save()
|
|
# logger.info(c1.to_dict())
|
|
# create_gtins_for_trading_goods('gtin_trading_goods_test.csv')
|
|
|
|
# LIVE FUNCTIONS BELOW
|
|
|
|
# find_articles_without_base_gtin()
|
|
|
|
# logger.info("Truncating GTINs")
|
|
# Article.clear_article_gtins()
|
|
logger.info("Creating new GTINs from base GTIN")
|
|
create_gtins(dry_run=False)
|
|
# logger.info("Creating new GTINs from trading goods CSV")
|
|
# create_gtins_for_trading_goods()
|
|
|
|
# logger.info("Update articles for batch management")
|
|
# set_storage_type()
|
|
|
|
# logger.info("Set zone information on article balance")
|
|
# set_zone_placement()
|
|
|
|
# logger.info("Updating alt units")
|
|
# update_decimals_on_alt_units()
|