# import generated xml import classes import csv from collections import defaultdict from datetime import date from itertools import chain from os.path import join from sqlite3 import IntegrityError from loguru import logger from sqlalchemy import update from sqlalchemy.orm import Session from src.config import settings from src.data.imports.schema.article_pt import ItemTypeSub as PartnerToolArticle, parse as parse_article_pt from src.data.imports.schema.article_variant_pt import ItemTypeSub as PartnerToolVariantArticle, \ parse as parse_variant_pt from src.data.imports.schema.article_b2b import ItemTypeSub as B2BArticle, parse as parse_article_b2b from src.data.imports.schema.pricelist import PriceListTypeSub as Import_PriceList, parse as parse_price_list # distinction shop schema from src.data.model.db.connection import get_sync_engine from src.data.model.schemas import Shop # import database schema from src.data.model.db.article import Season, ThemeGroup, Group, Color, Picture, Article, ArticleDetail, Pricelist from src.data.model.schemas.picture_types import PictureTypes # type definitions # since we have different classes for some imports from src.data.synchronisation.utils import create_lookup, extract_categorical_data, parse, merge_with_current_db_state Import_Article = PartnerToolArticle | B2BArticle Import_VariantArticle = PartnerToolVariantArticle | B2BArticle def import_article_data(session: Session): session.begin() # import b2b article_b2b_import = parse(settings.import_files_xml_article_b2b, parse_article_b2b) article_variant_b2b_import = parse(settings.import_files_xml_article_detail_b2b, parse_article_b2b) b2b_eans, b2b_colors, b2b_articles, b2b_images, _ = process_articles( article_b2b_import.Item, article_variant_b2b_import.Item, Shop.B2B ) # import partnertool article article_pt_import = parse(settings.import_files_xml_article_mobi, parse_article_pt) article_variant_pt_import = parse(settings.import_files_xml_article_detail_mobi, parse_variant_pt) pt_eans, pt_colors, pt_articles, pt_images, sales_stop_articles = process_articles( article_pt_import.Item, article_variant_pt_import.Item, Shop.PARTNERTOOL ) # import price lists price_lists_import = parse(settings.import_files_xml_price_list, parse_price_list) eans = b2b_eans | pt_eans price_lists = process_price_list(price_lists_import.PriceList, eans) # merge state merge_with_current_db_state( session, b2b_colors, b2b_articles, pt_colors, pt_articles, b2b_images, pt_images, # price_lists ) session.commit() # handle articles with sales stop by setting quantity of those to 0 handle_sales_stop_articles(session, sales_stop_articles) session.close() def process_articles( articles_imported: list[Import_Article], article_variants_imported: list[Import_VariantArticle], shop: Shop ): def get_quantity(quantity: float, variant, shop): unlimited_sell = bool(variant.SalesItems[-1].classification4) if shop.PARTNERTOOL and unlimited_sell: return 999 else: return quantity logger.info(f"importing articles from {str(shop)}") article_images: list[Picture] = [] color_images: set[Picture] = set() colors: set[Color] = set() eans: set[str] = set() sales_stop_articles: set[str] = set() articles: list[Article] = [] variants_by_article_lookup = create_lookup(key=lambda i: i.ReferenceItem.number, items=article_variants_imported) quantities = read_quantities(shop) for article_imported in articles_imported: if len(article_imported.Imp_ItemDeliveryDates) == 0: continue # parse images for image_type, image in process_images(article_imported): if image_type == PictureTypes.ARTICLE_IMAGE: article_images.append(image) else: color_images.add(image) # parse article delivery_date = article_imported.Imp_ItemDeliveryDates[0].deliveryDate article = Article( article_number=article_imported.number, composition=article_imported.Imp_MaterialComposition1.description, brand="1", delivery_date_from=delivery_date.dateFrom.date, delivery_date_to=delivery_date.dateUntil.date, group=Group(**extract_categorical_data(article_imported.Imp_ItemGroup)), season=Season(**extract_categorical_data(article_imported.Imp_Collection)), theme_group=ThemeGroup(**extract_categorical_data(article_imported.Imp_ItemThemeGroup)) ) # parse variants of article article.color_sizes = [] for variant in variants_by_article_lookup[article_imported.number]: if (shop == Shop.PARTNERTOOL and variant.imp_salesStop) or not variant.imp_activeColor: sales_stop_articles.add(variant.number) continue color = Color(**extract_categorical_data(variant.Imp_Color)) colors.add(color) ean = str(variant.Uoms[0].EANs[0].id) eans.add(ean) article.color_sizes.append( ArticleDetail( ean=ean, variant_number=variant.number, size=variant.Imp_Size.code, article_number=variant.ReferenceItem.number, # todo: eventually not needed size_register=variant.Imp_Size.SizeRegister.code, color_code=color.code, surcharge_code=variant.Imp_Size.surcharge, gender=variant.SalesItems[0].classification2, unlimited_sell=bool(variant.SalesItems[-1].classification4), quantity=get_quantity(quantities[variant.number], variant, shop) ) ) logger.info(f"article {article.article_number} was imported") articles.append(article) return eans, colors, articles, list(color_images) + article_images, sales_stop_articles def process_images(article: Import_Article) -> list[tuple[PictureTypes, Picture]]: def change_image_path(path: str, image_type: PictureTypes): """ Changes original image path to backend path according to type of image :param path: original path :param image_type: type of image :return: a new path of image """ file_name = path.split("/")[-1] return join(settings.pictures_path, str(image_type), file_name) return [ ( picture_type := PictureTypes.ARTICLE_IMAGE if "Fertigteile" in image.uri else PictureTypes.COLOR_IMAGE, Picture( type=picture_type, ref_id=article.number, path=change_image_path(image.uri, picture_type) ) ) for image in article.Imp_ItemColorImages ] def read_quantities(shop: Shop): # todo: dont really like the structure, but ok def to_float(value: str) -> float: return float(value.replace('.', '').replace(',', '.')) logger.info("importing quantities") data = defaultdict(float) if shop == Shop.B2B: path = join(settings.import_path, settings.import_files_csv_b2b_storage) with open(path, 'r') as csvfile: reader = csv.DictReader(csvfile, ['Fehlmenge', '', 'Gesamtbestand', '', 'Reservierte Menge', '', 'EAN']) for row in reader: value = to_float(row['Gesamtbestand']) - to_float(row['Fehlmenge']) - to_float(row['Reservierte Menge']) data.update({row['EAN']: value}) else: with open(join(settings.import_path, settings.import_files_csv_pt_open_amount), 'r') as csvfile: reader = csv.DictReader(csvfile, ['Soll-Menge', '', 'EAN']) for row in reader: data.setdefault(row['EAN'], to_float(row['Soll-Menge'])) with open(join(settings.import_path, settings.import_files_csv_pt_open_production), 'r') as csvfile: reader = csv.DictReader(csvfile, ['Fehlmenge', '', 'EAN']) for row in reader: ean = row['EAN'] if ean in data.keys(): data[ean] -= to_float(row['Fehlmenge']) else: data.setdefault(ean, -to_float(row['Fehlmenge'])) return data def process_price_list(price_lists_import: list[Import_PriceList], imported_eans: set[str]): logger.info("importing price lists") return [ Pricelist( code=price_list.code, price=price_definition.priceValue, article_detail_ean=ean ) for price_list in price_lists_import for price_definition in price_list.PriceDefinitions if (ean := price_definition.Item.eans[0].valueOf_) in imported_eans ] def handle_sales_stop_articles(session: Session, sales_stop_articles: set[str]): stmt = update(ArticleDetail).where(ArticleDetail.article_number.in_(sales_stop_articles)).values(quantity=0) session.execute(stmt) session.commit()