#!/usr/bin/env python3.6 import codecs import csv import logging import logging.config import os import shutil import time import openpyxl import pandas as pd import common def read_delete_ref(ref_path): # Expects ref_file to exist in ref folder filter_list = list() header_read = False filter_col_idx = -1 filter_val_idx = -1 with codecs.open( ref_path, "r", encoding='utf-8', errors='ignore') as f: reader = csv.reader(f, delimiter=';') for row in reader: if header_read: col_name = row[filter_col_idx].strip().lower().replace(" ", "_") filter_list.append((col_name, row[filter_val_idx])) else: if "Spalte" in row: filter_col_idx = row.index("Spalte") if "Delete" in row: filter_val_idx = row.index("Delete") if filter_col_idx < 0 or filter_val_idx < 0: raise Exception( "Not all columns found in {}".format(ref_path)) header_read = True return filter_list def read_ref(ref_path): # Expects ref_file to exist in ref folder filter_list = list() header_read = False filename_index = -1 filter1_column_index = -1 filter2_column_index = -1 filter1_value_index = -1 filter2_value_index = -1 base_price_column_index = -1 with codecs.open( ref_path, "r", encoding='utf-8', errors='ignore') as f: reader = csv.reader(f, delimiter=';') for row in reader: if header_read: filename = row[filename_index] filter1_column = row[filter1_column_index] filter1_value = row[filter1_value_index] filter2_column = row[filter2_column_index] filter2_value = row[filter2_value_index] base_price_column = row[base_price_column_index] multiplier_list = list() for idx, col_name in enumerate(price_multiplier_names): try: multiplier_value_txt = row[base_price_column_index + 1 + idx] multiplier_value = float( multiplier_value_txt.replace(",", ".")) except ValueError: logging.error("Failed to convert multiplier value %s", multiplier_value_txt) multiplier_value = 0.0 except Exception as ex: logging.error(ex) multiplier_value = 0.0 multiplier_list.append({ "MultiplierColumnName": col_name, "MultiplierValue": multiplier_value }) d = { "Filename": filename, "Filters": [], "BasePriceColumnName": base_price_column, "BasePriceMultipliers": multiplier_list } d["Filters"].append({ "FilterColumn": filter1_column, "FilterValue": filter1_value }) if filter2_column: d["Filters"].append({ "FilterColumn": filter2_column, "FilterValue": filter2_value }) filter_list.append(d) else: if "Datei" in row: filename_index = row.index("Datei") if "Filter1" in row: filter1_column_index = row.index("Filter1") if "Value1" in row: filter1_value_index = row.index("Value1") if "Filter2" in row: filter2_column_index = row.index("Filter2") if "Value2" in row: filter2_value_index = row.index("Value2") if "Preisgrundlage" in row: base_price_column_index = row.index("Preisgrundlage") if filename_index < 0 or filter1_column_index < 0 or \ filter1_value_index < 0 or filter2_column_index < 0 or \ filter2_value_index < 0 or base_price_column_index < 0: raise Exception( "Not all columns found in {}".format(ref_file)) price_multiplier_names = row[base_price_column_index + 1:] header_read = True return filter_list def create_offer_file(in_file, out_path, ref_multipliers, ref_deletes): df = pd.read_csv(in_file, sep=";") original_headers = df.columns df.columns = df.columns.str.strip().str.lower().str.replace(' ', '_') del_query = " & ".join(["{} != '{}'".format(p[0], p[1]) for p in ref_deletes]) logging.info("Delete query appplied: %s", del_query) df = df.query(del_query) df = df.reset_index(drop=True) for m in ref_multipliers: outfile_full = os.path.join(out_path, m["Filename"] + ".csv") if len(m["Filters"]) == 2: f1_col = "{}".format( m["Filters"][0]["FilterColumn"]).strip().lower().replace( " ", "_") f1_val = m["Filters"][0]["FilterValue"] f2_col = "{}".format( m["Filters"][1]["FilterColumn"]).strip().lower().replace( " ", "_") f2_val = m["Filters"][1]["FilterValue"] df_filtered = df[(df[f1_col] == f1_val) & (df[f2_col] == f2_val)] logging.info("Filtering '{}' by '{}' and '{}' by '{}'".format( f1_col, f1_val, f2_col, f2_val)) else: f1_col = "{}".format( m["Filters"][0]["FilterColumn"]).strip().lower().replace( " ", "_") f1_val = m["Filters"][0]["FilterValue"] df_filtered = df[df[f1_col] == f1_val] logging.info("Filtering '{}' by '{}'".format(f1_col, f1_val)) if df_filtered.shape[0] == 0: continue logging.info("Processing %s...", outfile_full) hdr_list = list() mult_list = list() for bpm in m["BasePriceMultipliers"]: hdr_list.append(bpm["MultiplierColumnName"]) mult_list.append(bpm["MultiplierValue"]) tl = [tuple(mult_list) for i in range(df_filtered.shape[0])] df_price_multiplier = pd.DataFrame.from_records(tl, columns=hdr_list) base_price_col_name = m["BasePriceColumnName"].strip().lower().replace( " ", "_") sales_price_list = df_filtered[base_price_col_name].tolist() if len(sales_price_list) != df_price_multiplier.shape[0]: raise Exception( "Price column length %d does not match number of df rows %d", len(sales_price_list), df_price_multiplier.shape[0]) # Drop the indexes for both dfs df_price_multiplier = df_price_multiplier.reset_index(drop=True) df_filtered = df_filtered.reset_index(drop=True) # Multiply each row by the sales price df_price_multiplier = df_price_multiplier.mul( sales_price_list, axis=0).round(2) if df_filtered.shape[0] != df_price_multiplier.shape[0]: raise Exception( "Number of price df %d rows does not match number of filtered df rows %d", df_price_multiplier.shape[0], df_filtered.shape[0]) # Restore the original header names df_filtered.columns = original_headers # Add the new columns to the existing dataframe df_filtered = df_filtered.join(df_price_multiplier) # writer = pd.ExcelWriter(outfile_full) # df_filtered.to_excel(writer, "Sheet1", index=False) # writer.save() df_filtered.to_csv(outfile_full, sep=";", index=False) def process_files(data_root): path_in = os.path.join(data_root, "in") path_out = os.path.join(data_root, "out") path_prc = os.path.join(data_root, "processed") path_ref = os.path.join(data_root, "ref") path_err = os.path.join(data_root, "error") path_log = os.path.join(data_root, "log") # Set up logging logging.config.dictConfig( common.get_logging_config( os.path.join(path_log, "process_matching_vendor_angebote.log"))) files = os.listdir(path_in) num_files = len(files) logging.info("Found {0} files...".format(num_files)) if num_files == 0: return full_ref_path = os.path.join(path_ref, "Ref_delete.csv") delete_ref = read_delete_ref(full_ref_path) full_ref_path = os.path.join(path_ref, "Ref_Vendor_Preise.csv") multiplier_ref = read_ref(full_ref_path) for f in files: in_file = os.path.join(path_in, f) proc_file = os.path.join(path_prc, f) err_file = os.path.join(path_err, f) if os.path.isfile(in_file): try: logging.debug("Processing {0}...".format(in_file)) create_offer_file(in_file, path_out, multiplier_ref, delete_ref) if os.path.isfile(in_file): shutil.move(in_file, proc_file) except Exception as e: logging.error(e) # Move in file to error shutil.move(in_file, err_file) def main(): data_root = r"/ftproot/" # data_root = r"/home/holger/git/bitbucket/thelen/data/" data_root = os.path.join(data_root, "Matching_Vendor_Angebote") process_files(data_root) if __name__ == "__main__": main()