import pandas as pd import gradio as gr import msoffcrypto import io import re import tempfile # Define transformation function def convert_ci(by_line_item_file, combined_file, password, max_rows): # Decrypt the protected "by line item" file decrypted = io.BytesIO() with open(by_line_item_file.name, "rb") as f: office_file = msoffcrypto.OfficeFile(f) office_file.load_key(password=password) office_file.decrypt(decrypted) # Load "by line item" data and skip last 3 rows by_line_item_df_raw = pd.read_excel(decrypted, skiprows=10, header=None, dtype= {4: str, 23: str}) by_line_item_df_raw = by_line_item_df_raw.iloc[:-3] # drop last 3 rows by_line_item_df = by_line_item_df_raw.iloc[:, [1, 4, 5, 6, 7, 8, 9, 10, 11, 23]].copy() # Drop fully empty rows by_line_item_df = by_line_item_df.dropna(how='all') # Clean up description for matching def clean_desc(row): desc_parts = [str(val).strip() for val in row.iloc[2:9] if pd.notna(val)] description = " ".join(desc_parts) description = re.sub(r"%", "", description) description = re.sub(r"\d+", "", description) return re.sub(r"[^\w\s]", "", description) by_line_item_df["Part"] = "" by_line_item_df["Tariff_Number"] = by_line_item_df.iloc[:, 1].apply(lambda x: str(x).replace('.0', '') if isinstance(x, float) and str(x).endswith('.0') else str(x)) by_line_item_df["Commercial_Description"] = by_line_item_df.apply(clean_desc, axis=1) by_line_item_df["MID_Code"] = by_line_item_df.iloc[:, 9].astype(str) by_line_item_df = by_line_item_df[["Part", "Tariff_Number", "Commercial_Description", "MID_Code"]] # Load the combined file and extract required fields invoice = pd.read_excel(combined_file.name, header=None) merged_value = invoice.iloc[8, 11] # L9 = 8th row, 11th col (0-indexed) mawb = invoice.iloc[8, 20] # U9 = 8th row, 20th col (0-indexed) invoice_no = str(merged_value) if pd.notna(merged_value) else "" combined_df_raw = pd.read_excel(combined_file.name, skiprows=10, header=None, dtype={10: str, 12: str}) combined_df_raw = combined_df_raw.iloc[:-3] # drop last 3 rows combined_df = combined_df_raw.iloc[:, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 12]].copy() combined_df.columns = [ "Tariff_Number", "Country_of_Origin", "Quantity", "Gross_Weight_KG", "Total_Line_Value", "Manufacturer_Name", "manufacturer_address", "Manufacturer_City", "Manufacturer_Country", "Manufacturer_Zip", "MID_Code" ] # Clean Tariff_Number and MID_Code for consistent merging combined_df["Tariff_Number"] = combined_df["Tariff_Number"].apply(lambda x: str(x).replace('.0', '') if isinstance(x, float) and str(x).endswith('.0') else str(x)) combined_df["MID_Code"] = combined_df["MID_Code"].astype(str) combined_df["Manufacturer_Zip"] = combined_df["Manufacturer_Zip"].astype(str).apply(lambda x: f"0{x}" if len(x) == 5 else x) # Format manufacturer address components def extract_address_parts(row): address = str(row["manufacturer_address"]) city = str(row["Manufacturer_City"]) address_1 = address state = "" if address.endswith("China"): parts = address.split(",") if len(parts) >= 2: state = parts[-2].replace(" Province", "").strip() elif "Singapore" in address: state = "" if city and city in address: last_occurrence_index = address.rfind(city) if last_occurrence_index != -1: address_1 = address[:last_occurrence_index].strip().rstrip(",") return pd.Series([state, address_1]) combined_df["Manufacturer_Country"] = combined_df["MID_Code"].str[:2] combined_df[["Manufacturer_State", "Manufacturer_Address_1"]] = combined_df.apply(extract_address_parts, axis=1) # Add calculated fields combined_df["Invoice_No"] = invoice_no combined_df["Country_of_Export"] = "CN" combined_df["Quantity_UOM"] = "pcs" combined_df["Buyer_Name"] = "SHEIN DISTRIBUTION CORPORATION" combined_df["Buyer_Address_1"] = "777 S. Alameda St" combined_df["Buyer_Address_2"] = "Suite 400" combined_df["Buyer_City"] = "Log Angeles" combined_df["Buyer_State"] = "CA" combined_df["Buyer_Zip"] = "90021" combined_df["Buyer_Country"] = "US" combined_df["Buyer_ID_Number"] = "" combined_df["Consignee_Name"] = "SHEIN DISTRIBUTION CORPORATION" combined_df["Consignee_Address_1"] = "777 S. Alameda St Suite" combined_df["Consignee_Address_2"] = "Suite 400" combined_df["Consignee_City"] = "Log Angeles" combined_df["Consignee_State"] = "CA" combined_df["Consignee_Zip"] = "90021" combined_df["Consignee_Country"] = "US" combined_df["Consignee_ID_Number"] = "" combined_df["Unit_Price"] = (combined_df["Total_Line_Value"] / combined_df["Quantity"]).round(2) combined_df["Net_Weight_KG"] = (combined_df["Gross_Weight_KG"] / combined_df["Quantity"]).round(2) combined_df["Gross_Weight_KG"] = combined_df["Gross_Weight_KG"].round(3) combined_df["Total_Line_Value"] = combined_df["Total_Line_Value"].round(2) # Drop duplicates in by_line_item_df for merge by_line_item_unique = by_line_item_df.drop_duplicates(subset=["Tariff_Number", "MID_Code"]) # Add empty columns empty_cols = [ 'SICountry', 'SP1', 'SP2', 'Zone_Status', 'Privileged_Filing_Date', 'Line_Piece_Count', 'ADD_Case_Number', 'CVD_Case_Number', 'AD_Non_Reimbursement_Statement', 'AD-CVD_Certification_Designation' ] for col in empty_cols: combined_df[col] = "" # Merge combined with by_line_item (first match by key) merged_df = combined_df.merge(by_line_item_unique, on=["Tariff_Number", "MID_Code"], how="left") column_order = [ 'Invoice_No', 'Part', 'Commercial_Description', 'Country_of_Origin', 'Country_of_Export', 'Tariff_Number', 'Quantity', 'Quantity_UOM', 'Unit_Price', 'Total_Line_Value', 'Net_Weight_KG', 'Gross_Weight_KG', 'Manufacturer_Name', 'Manufacturer_Address_1', 'Manufacturer_Address_2', 'Manufacturer_City', 'Manufacturer_State', 'Manufacturer_Zip', 'Manufacturer_Country', 'MID_Code', 'Buyer_Name', 'Buyer_Address_1', 'Buyer_Address_2', 'Buyer_City', 'Buyer_State', 'Buyer_Zip', 'Buyer_Country', 'Buyer_ID_Number', 'Consignee_Name', 'Consignee_Address_1', 'Consignee_Address_2', 'Consignee_City', 'Consignee_State', 'Consignee_Zip', 'Consignee_Country', 'Consignee_ID_Number', 'SICountry', 'SP1', 'SP2', 'Zone_Status', 'Privileged_Filing_Date', 'Line_Piece_Count', 'ADD_Case_Number', 'CVD_Case_Number', 'AD_Non_Reimbursement_Statement', 'AD-CVD_Certification_Designation' ] for col in column_order: if col not in merged_df.columns: merged_df[col] = "" merged_df = merged_df[column_order] # Replace the incorrect MID mapping_df = pd.read_csv("MID_replace.csv") mid_map = dict(zip(mapping_df["SHEIN_MID_Code"], mapping_df["Replacement"])) merged_df["MID_Code"] = merged_df["MID_Code"].apply(lambda x: re.sub(r"[^\w]", "", x)).map(mid_map).fillna(merged_df["MID_Code"]) # Replace HTS hts_mapping_df = pd.read_csv("HTS_replace.csv") hts_map = dict(zip(hts_mapping_df["old_hts"], hts_mapping_df["new_hts"])) merged_df["Tariff_Number"] = merged_df["Tariff_Number"].apply(lambda x: re.sub(r"[^\w]", "", x)).map(hts_map).fillna(merged_df["Tariff_Number"]) # Removed the special characters merged_df['Manufacturer_Name'] = merged_df['Manufacturer_Name'].apply(lambda x: re.sub(r"[^\w]", " ", x)) merged_df['Manufacturer_Address_1'] = merged_df['Manufacturer_Address_1'].apply(lambda x: re.sub(r"[^\w]", " ", x)) # Save merged output partitioned by user-defined max_rows outputs = [] for i, start in enumerate(range(0, len(merged_df), int(max_rows))): chunk = merged_df.iloc[start : start + int(max_rows)] suffix = chr(65 + i) if i < 26 else f"part{i + 1}" filename = f"{invoice_no}-{suffix}.xlsx" with pd.ExcelWriter(filename, engine='xlsxwriter') as writer: chunk.to_excel(writer, sheet_name="Converted", index=False) workbook = writer.book worksheet = writer.sheets['Converted'] # Format all cells as text text_format = workbook.add_format({'num_format': '@'}) worksheet.set_column(0, len(chunk.columns) - 1, None, text_format) outputs.append(filename) import zipfile zip_name = f"{mawb} T01 Manifest.zip" with zipfile.ZipFile(zip_name, 'w') as zipf: for file in outputs: zipf.write(file, arcname=file) return zip_name # Gradio Interface # Gradio Interface iface = gr.Interface( fn=convert_ci, inputs=[ gr.File(label="Upload 'By Line Item' File (.xlsx)", type="filepath"), gr.File(label="Upload 'Combined' File (.xlsx)", type="filepath"), gr.Textbox(label="Password for 'By Line Item' File", type="password"), gr.Number(label="Max Rows per Output File", value=998, precision=0) ], outputs=gr.File(label="Download Zipped Excel Output", file_types=[".zip"]), title="SHEIN CI to Magaya Format Conversion Tool", description="Upload both 'By Line Item' (with password) and 'Combined' Excel files to merge and generate the Manifest in Magaya T01 Format." ) if __name__ == "__main__": iface.launch()