Spaces:
Sleeping
Sleeping
| import streamlit as st | |
| import pandas as pd | |
| import json | |
| from datetime import datetime | |
| import io | |
| import os | |
| import pytz | |
| # --- Inizializzazione Session State --- | |
| if 'file_uploader_key_json' not in st.session_state: | |
| st.session_state.file_uploader_key_json = 'json_uploader_0' | |
| if 'file_uploader_key_customers' not in st.session_state: | |
| st.session_state.file_uploader_key_customers = 'customers_uploader_0' | |
| if 'customer_excel_loaded' not in st.session_state: | |
| st.session_state.customer_excel_loaded = False | |
| if 'customers_df' not in st.session_state: | |
| st.session_state.customers_df = pd.DataFrame() | |
| # --- CSS personalizzato --- | |
| st.markdown( | |
| """ | |
| <style> | |
| .stApp { | |
| background-color: #FFFFCC; /* Giallo pastello */ | |
| background-attachment: fixed; | |
| } | |
| h1, h2 { | |
| color: #1E90FF; /* Blu acceso */ | |
| } | |
| p, label { | |
| color: #000080; /* Blu scuro */ | |
| } | |
| .stButton>button { | |
| color: white; | |
| background-color: #1E90FF; /* Blu acceso */ | |
| border: none; | |
| border-radius: 5px; | |
| padding: 10px 20px; | |
| font-size: 16px; | |
| } | |
| .stButton>button:hover { | |
| background-color: #4169E1; /* Blu medio al passaggio del mouse */ | |
| } | |
| </style> | |
| """, | |
| unsafe_allow_html=True | |
| ) | |
| # Definizione del fuso orario (per coerenza, anche se le date Outbound sembrano solo data) | |
| ITALY_TIMEZONE = pytz.timezone("Europe/Rome") | |
| UTC_TIMEZONE = pytz.utc | |
| # Funzione di conversione per Outbound | |
| def convert_json_to_excel_data_outbound(json_file_content: str, customers_df_input: pd.DataFrame = None): | |
| try: | |
| json_data = json.loads(json_file_content) | |
| # Campi a livello radice | |
| site_code_json = json_data.get('siteCode') | |
| company_code_json = json_data.get('companyCode') | |
| # Campi sotto 'outbound' | |
| outbound_info = json_data.get('outbound', {}) | |
| order_number = outbound_info.get('orderNumber') | |
| order_type_json = outbound_info.get('orderType') | |
| delivery_type_fixed = "DELIVERY" | |
| order_doc_date_str = outbound_info.get('orderDocDate') | |
| date_detail_json = outbound_info.get('orderDateDetail') | |
| due_date_str = outbound_info.get('orderDate') | |
| reason_code = outbound_info.get('reasonCode') | |
| payment_info_value = outbound_info.get('paymentInfo', {}).get('value') | |
| delivery_customer = outbound_info.get('deliveryCustomer', {}) | |
| customer_code = delivery_customer.get('code') | |
| customer_mail = delivery_customer.get('refMail') | |
| # Consignee data (from JSON) | |
| consignee_name = delivery_customer.get('name') | |
| consignee_address_base = delivery_customer.get('address', '') | |
| consignee_address_number = delivery_customer.get('addressNumber', '') | |
| consignee_address = f"{consignee_address_base} {consignee_address_number}".strip() if consignee_address_base or consignee_address_number else "" | |
| consignee_city = delivery_customer.get('city') | |
| consignee_province = delivery_customer.get('province') | |
| consignee_zip_code = delivery_customer.get('zipCode') | |
| consignee_country = delivery_customer.get('country') | |
| # Customer data (from Excel lookup) | |
| customer_name = "" | |
| customer_address = "" | |
| customer_city = "" | |
| customer_province = "" | |
| customer_zip_code = "" | |
| customer_country = "" | |
| # Esegui la lookup solo se customers_df_input è fornito e non vuoto | |
| if customers_df_input is not None and not customers_df_input.empty and customer_code and company_code_json: | |
| # Filtra per 'code' e 'company.code' (assicurandosi che siano stringhe) | |
| # Aggiunto controllo per tipo stringa prima del confronto | |
| customer_data = customers_df_input[ | |
| (customers_df_input['code'].astype(str) == str(customer_code)) & | |
| (customers_df_input['company.code'].astype(str) == str(company_code_json)) | |
| ] | |
| if not customer_data.empty: | |
| customer_row = customer_data.iloc[0] | |
| customer_name = customer_row.get('partner.name', '') | |
| excel_customer_address_base = customer_row.get('partner.address.address', '') | |
| excel_customer_address_number = customer_row.get('partner.address.addressNumber', '') | |
| customer_address = f"{excel_customer_address_base} {excel_customer_address_number}".strip() if excel_customer_address_base or excel_customer_address_number else "" | |
| customer_city = customer_row.get('partner.address.city', '') | |
| customer_province = customer_row.get('partner.address.province', '') | |
| customer_zip_code = customer_row.get('partner.address.zipCode', '') | |
| customer_country = customer_row.get('partner.address.country', '') | |
| else: | |
| # Errore bloccante se l'anagrafica non viene trovata | |
| st.error(f"ERRORE: Manca anagrafica per customer.code '{customer_code}' e company.code '{company_code_json}' nel file Excel dei clienti. Impossibile procedere con questo JSON.") | |
| return None # Ritorna None per indicare un errore bloccante | |
| elif not customer_code or not company_code_json: | |
| st.warning(f"Attenzione: Customer Code o Company Code mancante nel JSON per l'ordine '{order_number}'. Impossibile effettuare la lookup sull'Excel.") | |
| elif customers_df_input is None or customers_df_input.empty: | |
| st.error("ERRORE: Il file Excel dell'anagrafica partner non è stato caricato o è vuoto. Caricare prima l'anagrafica.") | |
| return None # Errore bloccante se l'Excel clienti non è disponibile | |
| # Note fields | |
| note_ddt = outbound_info.get('noteDdt') | |
| note_warehouse = outbound_info.get('noteWarehouse') | |
| note_delivery = outbound_info.get('noteDelivery') | |
| note_shipment = outbound_info.get('noteShipment') | |
| reason_code_shp = outbound_info.get('reasonCodeShp') | |
| info_delivery = outbound_info.get('infoDelivery') | |
| booking_info = outbound_info.get('bookingInfo') | |
| # Formattazione delle date | |
| order_date_formatted = None | |
| if order_doc_date_str: | |
| try: | |
| dt_object = datetime.fromisoformat(order_doc_date_str.replace('Z', '')) | |
| order_date_formatted = dt_object.strftime('%d/%m/%Y') | |
| except ValueError: | |
| order_date_formatted = order_doc_date_str | |
| st.warning(f"Attenzione: Impossibile parsare 'orderDocDate' '{order_doc_date_str}'. Verrà usata la stringa originale.") | |
| due_date_formatted = None | |
| if due_date_str: | |
| try: | |
| dt_object = datetime.fromisoformat(due_date_str.replace('Z', '')) | |
| due_date_formatted = dt_object.strftime('%d/%m/%Y') | |
| except ValueError: | |
| due_date_formatted = due_date_str | |
| st.warning(f"Attenzione: Impossibile parsare 'orderDate' (dueDate) '{due_date_str}'. Verrà usata la stringa originale.") | |
| details = outbound_info.get('details', []) | |
| processed_data_for_this_file = [] | |
| for item in details: | |
| row = { | |
| 'siteCode': site_code_json, | |
| 'companyCode': company_code_json, | |
| 'orderNumber': order_number, | |
| 'orderType': order_type_json, | |
| 'deliveryType': delivery_type_fixed, | |
| 'orderDate': order_date_formatted, | |
| 'dateDetail': date_detail_json, | |
| 'dueDate': due_date_formatted, | |
| 'reasonCode': reason_code, | |
| 'paymentValue': payment_info_value, | |
| 'customer.code': customer_code, | |
| 'customer.name': customer_name, | |
| 'customer.address': customer_address, | |
| 'customer.city': customer_city, | |
| 'customer.province': customer_province, | |
| 'customer.zipCode': customer_zip_code, | |
| 'customer.country': customer_country, | |
| 'consignee.name': consignee_name, | |
| 'consignee.address': consignee_address, | |
| 'consignee.city': consignee_city, | |
| 'consignee.province': consignee_province, | |
| 'consignee.zipCode': consignee_zip_code, | |
| 'consignee.country': consignee_country, | |
| 'customer.mail': customer_mail, | |
| 'details.positionId': item.get('positionId'), | |
| 'details.itemNumber': item.get('itemCode'), | |
| 'details.requestedQt': item.get('quantity'), | |
| 'details.lotId': item.get('lotId'), | |
| 'details.stockOwner': item.get('stockOwner'), | |
| 'details.magCli': item.get('warehouse'), | |
| 'details.availability': item.get('availability'), | |
| 'details.accountingStatus': item.get('accountingStatus'), | |
| 'details.refOrder': item.get('refOrder'), | |
| 'noteDDT': note_ddt, | |
| 'noteMag': note_warehouse, | |
| 'noteDelivery': note_delivery, | |
| 'noteShipment': note_shipment, | |
| 'reasonCodeShp': reason_code_shp, | |
| 'infoDelivery': info_delivery, | |
| 'bookingInfo': booking_info | |
| } | |
| processed_data_for_this_file.append(row) | |
| if not processed_data_for_this_file: | |
| return pd.DataFrame() | |
| return pd.DataFrame(processed_data_for_this_file) | |
| except json.JSONDecodeError as e: | |
| st.error(f"Errore nel file JSON: {e}. Assicurati che il contenuto sia un JSON valido.") | |
| return None | |
| except KeyError as ke: | |
| st.error(f"Errore: Chiave JSON mancante o struttura inattesa: {ke}. Assicurati che il JSON abbia la struttura attesa.") | |
| return None | |
| except Exception as e: | |
| st.error(f"Si è verificato un errore generico durante la conversione di un file JSON: {e}") | |
| return None | |
| # --- Interfaccia Streamlit --- | |
| st.title("Convertitore JSON Outbound a Template Excel Outbound") | |
| st.write("Questo strumento converte uno o più file JSON di ordini Outbound in un unico foglio Excel formattato, utilizzando un'anagrafica partner esterna.") | |
| st.header("1. Carica Excel con Anagrafica Partner") | |
| uploaded_customers_file = st.file_uploader( | |
| "Scegli il file Excel con l'anagrafica dei partner", | |
| type=["xlsx", "xls"], | |
| accept_multiple_files=False, | |
| key=st.session_state.file_uploader_key_customers | |
| ) | |
| # Inizializza customers_df per il contesto corrente del rerun | |
| customers_df = st.session_state.customers_df | |
| if uploaded_customers_file: | |
| try: | |
| # Leggi l'Excel partendo dalla riga 2 (indice 1) per le intestazioni | |
| temp_customers_df = pd.read_excel(uploaded_customers_file, header=1) | |
| # Colonne richieste nell'Excel clienti | |
| required_excel_cols = [ | |
| 'code', 'company.code', 'partner.name', 'partner.address.address', | |
| 'partner.address.addressNumber', 'partner.address.city', | |
| 'partner.address.province', 'partner.address.zipCode', 'partner.address.country' | |
| ] | |
| # Verifica che tutte le colonne richieste siano presenti | |
| missing_cols = [col for col in required_excel_cols if col not in temp_customers_df.columns] | |
| if missing_cols: | |
| st.error(f"Errore: Il file Excel dell'anagrafica partner manca le seguenti colonne essenziali: {', '.join(missing_cols)}. Caricare un file corretto.") | |
| st.session_state.customer_excel_loaded = False | |
| st.session_state.customers_df = pd.DataFrame() # Resetta il DF in caso di errore | |
| else: | |
| # Assicurati che le colonne per il merge siano stringhe per evitare problemi di tipo | |
| temp_customers_df['code'] = temp_customers_df['code'].astype(str) | |
| temp_customers_df['company.code'] = temp_customers_df['company.code'].astype(str) | |
| st.session_state.customers_df = temp_customers_df # Salva il DF nello session_state | |
| st.session_state.customer_excel_loaded = True | |
| st.success("File Excel Anagrafica Partner caricato con successo e verificato.") | |
| customers_df = st.session_state.customers_df # Aggiorna la variabile locale | |
| except Exception as e: | |
| st.error(f"Errore nel caricamento del file Excel Anagrafica Partner: {e}. Assicurati che sia un file Excel valido e che le intestazioni siano sulla seconda riga.") | |
| st.session_state.customer_excel_loaded = False | |
| st.session_state.customers_df = pd.DataFrame() # Resetta il DF in caso di errore | |
| # Aggiorna la variabile locale customers_df ogni volta che il session_state cambia | |
| customers_df = st.session_state.customers_df | |
| st.header("2. Carica i tuoi file JSON degli Ordini Outbound") | |
| uploaded_json_files = st.file_uploader( | |
| "Scegli uno o più file JSON di tipo 'Ordine_Outbound'", | |
| type=["json"], | |
| accept_multiple_files=True, | |
| key=st.session_state.file_uploader_key_json, | |
| disabled=not st.session_state.customer_excel_loaded # Disabilita se l'Excel non è stato caricato | |
| ) | |
| all_processed_data = [] | |
| # Flag per tracciare se c'è stato un errore bloccante nella conversione di qualsiasi JSON | |
| has_blocking_error = False | |
| if uploaded_json_files and st.session_state.customer_excel_loaded: | |
| st.info(f"Caricati {len(uploaded_json_files)} file JSON. Elaborazione in corso...") | |
| for uploaded_file in uploaded_json_files: | |
| file_name = uploaded_file.name | |
| if not file_name.lower().startswith("ordine_outbound"): | |
| st.error(f"Errore: Il file '{file_name}' non inizia con 'Ordine_Outbound'. Verrà ignorato.") | |
| has_blocking_error = True | |
| continue | |
| try: | |
| json_content_bytes = uploaded_file.read() | |
| json_content_string = json_content_bytes.decode('utf-8') | |
| # Passa customers_df dallo session state alla funzione di conversione | |
| df_this_file = convert_json_to_excel_data_outbound(json_content_string, st.session_state.customers_df) | |
| if df_this_file is not None and not df_this_file.empty: | |
| all_processed_data.append(df_this_file) | |
| elif df_this_file is None: # Se la funzione ritorna None, c'è stato un errore bloccante | |
| has_blocking_error = True | |
| # Il messaggio di errore specifico è già gestito all'interno di convert_json_to_excel_data_outbound | |
| except Exception as e: | |
| st.error(f"Si è verificato un errore imprevisto durante la lettura o l'elaborazione del file '{file_name}': {e}") | |
| has_blocking_error = True | |
| if not has_blocking_error and all_processed_data: | |
| final_df = pd.concat(all_processed_data, ignore_index=True) | |
| st.success(f"Elaborazione di {len(all_processed_data)} file JSON completata! Pronto per il download.") | |
| # --- Post-processing per la formattazione Excel --- | |
| # Definizione delle colonne di testata | |
| header_columns = [ | |
| 'siteCode', 'companyCode', 'orderNumber', 'orderType', 'deliveryType', | |
| 'orderDate', 'dateDetail', 'dueDate', 'reasonCode', | |
| 'paymentValue', | |
| 'customer.code', 'customer.name', 'customer.address', 'customer.city', | |
| 'customer.province', 'customer.zipCode', 'customer.country', | |
| 'consignee.name', 'consignee.address', 'consignee.city', 'consignee.province', | |
| 'consignee.zipCode', 'consignee.country', 'customer.mail', | |
| 'noteDDT', 'noteMag', 'noteDelivery', 'noteShipment', | |
| 'reasonCodeShp', 'infoDelivery', 'bookingInfo' | |
| ] | |
| # Ordina il DataFrame per orderNumber e details.positionId per raggruppare gli ordini | |
| final_df = final_df.sort_values(by=['orderNumber', 'details.positionId']).reset_index(drop=True) | |
| # Crea una maschera booleana per identificare le righe duplicate per orderNumber | |
| rows_to_clear_headers = final_df.duplicated(subset=['orderNumber'], keep='first') | |
| # Azzerare le colonne di testata per le righe identificate | |
| final_df.loc[rows_to_clear_headers, header_columns] = "" | |
| # --- Creazione file Excel --- | |
| output = io.BytesIO() | |
| writer = pd.ExcelWriter(output, engine='xlsxwriter') | |
| sheet_name_excel = 'Delivery' # Nome del foglio modificato | |
| workbook = writer.book | |
| worksheet = workbook.add_worksheet(sheet_name_excel) | |
| # Lista delle colonne finali nell'ordine desiderato | |
| final_columns_order = [ | |
| 'siteCode', 'companyCode', 'orderNumber', 'orderType', 'deliveryType', | |
| 'orderDate', 'dateDetail', 'dueDate', 'reasonCode', | |
| 'paymentValue', | |
| 'customer.code', 'customer.name', 'customer.address', 'customer.city', | |
| 'customer.province', 'customer.zipCode', 'customer.country', | |
| 'consignee.name', 'consignee.address', 'consignee.city', 'consignee.province', | |
| 'consignee.zipCode', 'consignee.country', 'customer.mail', | |
| 'details.positionId', 'details.itemNumber', 'details.requestedQt', | |
| 'details.lotId', 'details.stockOwner', 'details.magCli', 'details.availability', | |
| 'details.accountingStatus', 'details.refOrder', | |
| 'noteDDT', 'noteMag', 'noteDelivery', 'noteShipment', | |
| 'reasonCodeShp', 'infoDelivery', 'bookingInfo' | |
| ] | |
| # Assicurati che il DataFrame abbia tutte le colonne nell'ordine desiderato | |
| for col in final_columns_order: | |
| if col not in final_df.columns: | |
| final_df[col] = "" | |
| final_df = final_df[final_columns_order] | |
| # Larghezze delle colonne aggiornate | |
| col_widths = { | |
| 'siteCode': 10, 'companyCode': 15, 'orderNumber': 15, 'orderType': 10, 'deliveryType': 12, | |
| 'orderDate': 12, 'dateDetail': 15, 'dueDate': 12, 'reasonCode': 12, | |
| 'paymentValue': 15, | |
| 'customer.code': 15, 'customer.name': 30, 'customer.address': 30, 'customer.city': 15, | |
| 'customer.province': 10, 'customer.zipCode': 10, 'customer.country': 10, | |
| 'consignee.name': 30, 'consignee.address': 30, 'consignee.city': 15, 'consignee.province': 10, | |
| 'consignee.zipCode': 10, 'consignee.country': 10, 'customer.mail': 25, | |
| 'details.positionId': 12, 'details.itemNumber': 20, 'details.requestedQt': 15, | |
| 'details.lotId': 10, 'details.stockOwner': 15, 'details.magCli': 15, 'details.availability': 15, | |
| 'details.accountingStatus': 18, 'details.refOrder': 15, | |
| 'noteDDT': 20, 'noteMag': 20, 'noteDelivery': 20, 'noteShipment': 20, | |
| 'reasonCodeShp': 15, 'infoDelivery': 20, 'bookingInfo': 20 | |
| } | |
| # Formato data per le colonne di data | |
| date_format = workbook.add_format({'num_format': 'dd/mm/yyyy'}) | |
| # Scrivi i nomi delle colonne nella riga 2 (indice 1 in xlsxwriter) | |
| for col_num, col_name in enumerate(final_df.columns.values): | |
| worksheet.write(1, col_num, col_name) | |
| # Imposta le larghezze delle colonne | |
| if col_name in col_widths: | |
| worksheet.set_column(col_num, col_num, col_widths[col_name]) | |
| else: | |
| worksheet.set_column(col_num, col_num, 15) | |
| # Applica il formato data alle colonne 'orderDate' e 'dueDate' | |
| if col_name in ['orderDate', 'dueDate']: | |
| worksheet.set_column(col_num, col_num, col_widths[col_name], date_format) | |
| # Scrivi i dati a partire dalla riga 3 (indice 2 in xlsxwriter) | |
| final_df.to_excel(writer, sheet_name=sheet_name_excel, index=False, header=False, startrow=2) | |
| writer.close() | |
| output.seek(0) | |
| timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") | |
| output_file_name = f"DeliveryOrders_{timestamp}.xlsx" # Nome del file di output | |
| st.download_button( | |
| label="Scarica file Excel Unificato", | |
| data=output.getvalue(), | |
| file_name=output_file_name, | |
| mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" | |
| ) | |
| elif uploaded_json_files and has_blocking_error: | |
| st.error("Si sono verificati errori bloccanti durante l'elaborazione dei JSON. Correggere gli errori per generare il file.") | |
| elif uploaded_json_files and not st.session_state.customer_excel_loaded: | |
| st.warning("Per elaborare i file JSON, devi prima caricare e validare un file Excel con l'anagrafica partner.") | |
| else: | |
| st.info("Carica i tuoi file per iniziare.") | |
| # --- Funzione per il reset --- | |
| def clear_inputs(): | |
| st.session_state.file_uploader_key_json = f'json_uploader_{datetime.now().timestamp()}' | |
| st.session_state.file_uploader_key_customers = f'customers_uploader_{datetime.now().timestamp() + 1}' # Assicurati che le chiavi siano diverse | |
| st.session_state.customer_excel_loaded = False | |
| st.session_state.customers_df = pd.DataFrame() # Resetta il DataFrame dell'anagrafica | |
| st.markdown("---") | |
| st.button("Pulisci Input (RESET)", on_click=clear_inputs) | |
| st.markdown("Sviluppato con Streamlit e Python.") |