Update app.py

app.py

@@ -2,13 +2,34 @@ import gradio as gr
 import pandas as pd
 import re
 import unicodedata
-import io
 import tempfile
 
-# ---------- Helper Functions ----------
-
+# ---------- UPS TARGET COLUMN ORDER (NO HEADER) ----------
+TARGET_COLUMNS = [
+    "Contact Name","Company or Name","Country","Address 1","Address 2","Address 3","City",
+    "State/Prov/Other","Postal Code","Telephone","Ext","Residential Ind","Consignee Email",
+    "Packaging Type","Customs Value","Weight","Length","Width","Height","Unit of Measure",
+    "Description of Goods","Documents of No Commercial Value","GNIFC","Pkg Decl Value",
+    "Service","Delivery Confirm","Shipper Release","Ret of Documents","Saturday Deliver",
+    "Carbon Neutral","Large Package","Addl handling","Reference 1","Reference 2","Reference 3",
+    "QV Notif 1-Addr","QV Notif 1-Ship","QV Notif 1-Excp","QV Notif 1-Delv",
+    "QV Notif 2-Addr","QV Notif 2-Ship","QV Notif 2-Excp","QV Notif 2-Delv",
+    "QV Notif 3-Addr","QV Notif 3-Ship","QV Notif 3-Excp","QV Notif 3-Delv",
+    "QV Notif 4-Addr","QV Notif 4-Ship","QV Notif 4-Excp","QV Notif 4-Delv",
+    "QV Notif 5-Addr","QV Notif 5-Ship","QV Notif 5-Excp","QV Notif 5-Delv",
+    "QV Notif Msg","QV Failure Addr","UPS Premium Care","ADL Location ID","ADL Media Type",
+    "ADL Language","ADL Notification Addr","ADL Failure Addr","ADL COD Value",
+    "ADL Deliver to Addressee","ADL Shipper Media Type","ADL Shipper Language",
+    "ADL Shipper Notification Addr","ADL Direct Delivery Only",
+    "Electronic Package Release Authentication","Lithium Ion Alone","Lithium Ion In Equipment",
+    "Lithium Ion With_Equipment","Lithium Metal Alone","Lithium Metal In Equipment",
+    "Lithium Metal With Equipment","Weekend Commercial Delivery","Dry Ice Weight",
+    "Merchandise Description","UPS Ground Saver Limited Quantity/Lithium Battery"
+]
+
+# ---------- HELPERS ----------
 def clean_text(s: str) -> str:
-    """Remove unwanted characters and normalize."""
+    """Remove 'ÿ', control chars and normalize to printable ASCII."""
     if pd.isna(s):
         return ""
     s = str(s).replace("ÿ", "")
@@ -16,63 +37,81 @@ def clean_text(s: str) -> str:
     s = "".join(ch for ch in s if 32 <= ord(ch) <= 126)
     return s.strip()
 
-def format_zip(zip_code):
-    """Pad ZIP codes to 5 digits."""
+def format_zip(zip_code) -> str:
+    """Pad to 5 digits; strip non-digits first."""
     if pd.isna(zip_code):
         return ""
-    z = str(zip_code).strip()
-    z = re.sub(r"[^\d]", "", z)
+    z = re.sub(r"[^\d]", "", str(zip_code).strip())
     if not z:
         return ""
     return z.zfill(5)[:5]
 
 def flow_address_lines(lines, maxlen=35, maxlines=3):
-    """Split long address lines into multiple lines."""
+    """Word-aware wrap into up to 3 lines, hard-splitting very long tokens."""
     tokens = []
     for ln in lines:
         txt = clean_text(ln)
         if txt:
             tokens.extend(txt.split())
     out = ["", "", ""]
-    line_i = 0
+    i = 0
     for tok in tokens:
         while len(tok) > maxlen:
             chunk, tok = tok[:maxlen], tok[maxlen:]
-            if line_i >= maxlines:
-                return out
-            if out[line_i]:
-                line_i += 1
-                if line_i >= maxlines:
-                    return out
-            out[line_i] = chunk
-            line_i += 1
-            if line_i >= maxlines:
-                return out
-        if line_i >= maxlines:
-            return out
-        add_len = len(tok) if not out[line_i] else len(tok) + 1
-        if len(out[line_i]) + add_len <= maxlen:
-            out[line_i] = (out[line_i] + (" " if out[line_i] else "") + tok).strip()
+            if i >= maxlines:
+                return [s[:maxlen] for s in out]
+            if out[i]:
+                i += 1
+                if i >= maxlines:
+                    return [s[:maxlen] for s in out]
+            out[i] = chunk
+            i += 1
+            if i >= maxlines:
+                return [s[:maxlen] for s in out]
+        if i >= maxlines:
+            return [s[:maxlen] for s in out]
+        add_len = len(tok) if not out[i] else len(tok) + 1
+        if len(out[i]) + add_len <= maxlen:
+            out[i] = (out[i] + (" " if out[i] else "") + tok).strip()
         else:
-            line_i += 1
-            if line_i >= maxlines:
-                return out
-            out[line_i] = tok
-    return [ln[:maxlen] for ln in out]
-
-def convert_dry_ice_kg(x):
-    """Convert lbs → kg and round."""
-    if pd.isna(x) or str(x).strip() == "":
-        return ""
-    try:
-        kg = round(float(str(x).strip()) / 2.2)
-        return str(int(kg))
-    except:
-        return ""
-
-# ---------- Main Cleaning Function ----------
-
-def clean_csv(file):
+            i += 1
+            if i >= maxlines:
+                return [s[:maxlen] for s in out]
+            out[i] = tok
+    return [s[:maxlen] for s in out]
+
+def to_str_series(df, colname):
+    """Return a cleaned string Series for an existing column, else blanks."""
+    if colname in df.columns:
+        return df[colname].apply(lambda x: clean_text(x))
+    return pd.Series([""] * len(df))
+
+def to_num_str_series(df, colname):
+    """Return numeric-looking strings (or blanks) for an existing column."""
+    if colname in df.columns:
+        return df[colname].apply(lambda x: "" if pd.isna(x) or str(x).strip()=="" else str(x).strip())
+    return pd.Series([""] * len(df))
+
+def dry_ice_lbs_to_kg_str(df, colname):
+    if colname in df.columns:
+        def conv(x):
+            if pd.isna(x) or str(x).strip()=="":
+                return ""
+            try:
+                return str(int(round(float(str(x).strip())/2.2)))
+            except:
+                return ""
+        return df[colname].apply(conv)
+    return pd.Series([""] * len(df))
+
+def zip_series(df, colname):
+    if colname in df.columns:
+        return df[colname].apply(format_zip)
+    return pd.Series([""] * len(df))
+
+# ---------- CORE PROCESS ----------
+def build_ups_batch_no_header(file):
+    # Load CSV with fallback encodings
     try:
         df = pd.read_csv(file.name, encoding="latin1")
     except Exception:
@@ -80,58 +119,87 @@ def clean_csv(file):
 
     df.columns = df.columns.str.strip()
 
-    # ZIP correction
-    if "ZipCode" in df.columns:
-        df["ZipCode"] = df["ZipCode"].map(format_zip)
-
-    # Split address lines
-    addr1, addr2, addr3 = [], [], []
+    # Address wrap (≤35 chars each)
+    a1_list, a2_list, a3_list = [], [], []
     for _, row in df.iterrows():
         a1, a2, a3 = flow_address_lines([
-            row.get("Address1", ""), row.get("Address2", ""), row.get("Address3", "")
+            row.get("Address1",""), row.get("Address2",""), row.get("Address3","")
         ])
-        addr1.append(a1)
-        addr2.append(a2)
-        addr3.append(a3)
-    df["Address1"] = addr1
-    df["Address2"] = addr2
-    df["Address3"] = addr3
-
-    # Clean key fields
-    for col in ["Company Name", "Contact Name", "City", "State", "Phone Number", "Email"]:
-        if col in df.columns:
-            df[col] = df[col].map(clean_text)
-
-    # Convert Dry Ice Weight
-    if "Dry Ice Weight" in df.columns:
-        df["Dry Ice Weight (kg)"] = df["Dry Ice Weight"].map(convert_dry_ice_kg)
-
-    # Save to a temp file for download
-    temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".csv")
-    df.to_csv(temp_file.name, index=False, encoding="utf-8-sig")
-    temp_file.close()
-
-    return temp_file.name  # ✅ Return single file path
-
-# ---------- Gradio UI ----------
-
-title = "UPS Shipment CSV Cleaner"
-description = """
-Upload your **raw shipment CSV file**.  
-The tool will:
-- Remove bad characters (e.g. ÿ)
-- Pad ZIP codes to 5 digits  
-- Split long addresses into 35-character lines  
-- Convert Dry Ice Weight (lbs → kg)
-Then download your cleaned CSV ready for UPS Batch import.
-"""
+        a1_list.append(a1); a2_list.append(a2); a3_list.append(a3)
+
+    # Build output strictly in TARGET_COLUMNS order
+    out = pd.DataFrame({c: [""] * len(df) for c in TARGET_COLUMNS})
+
+    # Required / mapped fields
+    out["Contact Name"]          = to_str_series(df, "Contact Name")
+    out["Company or Name"]       = to_str_series(df, "Company Name")
+    out["Country"]               = "US"
+    out["Address 1"]             = pd.Series(a1_list)
+    out["Address 2"]             = pd.Series(a2_list)
+    out["Address 3"]             = pd.Series(a3_list)
+    out["City"]                  = to_str_series(df, "City")
+    out["State/Prov/Other"]      = to_str_series(df, "State")
+    out["Postal Code"]           = zip_series(df, "ZipCode")
+    out["Telephone"]             = to_str_series(df, "Phone Number")
+    out["Consignee Email"]       = to_str_series(df, "Email")
+
+    # Dimensions / weight
+    out["Weight"] = to_num_str_series(df, "Weight")
+    out["Length"] = to_num_str_series(df, "Length")
+    out["Width"]  = to_num_str_series(df, "Width")
+    out["Height"] = to_num_str_series(df, "Height")
+
+    # Fixed UPS details per your rules
+    out["Packaging Type"]        = "2"      # not "02"
+    out["Service"]               = "01"     # include leading zero
+    out["Delivery Confirm"]      = "S"
+    out["Description of Goods"]  = "Dry Ice Biological Shipment"
+    out["Merchandise Description"]= "Dry Ice Biological Shipment"
+    out["ADL Language"]          = ""       # blank
+
+    # Dry ice conversion (lbs -> kg, rounded)
+    out["Dry Ice Weight"]        = dry_ice_lbs_to_kg_str(df, "Dry Ice Weight")
+
+    # References mapping
+    out["Reference 1"]           = to_num_str_series(df, "PO Number")
+    out["Reference 2"]           = to_num_str_series(df, "Invoice Number")
+    out["Reference 3"]           = to_num_str_series(df, "Customer Reference")
+
+    # QV Notification flags/addresses
+    out["QV Notif 1-Addr"]       = to_str_series(df, "Email")  # recipient email
+    out["QV Notif 1-Ship"]       = "1"
+    out["QV Notif 1-Excp"]       = "1"
+    out["QV Notif 1-Delv"]       = "1"
+
+    out["QV Notif 2-Addr"]       = "shaqdong@apexglobe.com"
+    out["QV Notif 2-Ship"]       = "1"
+    out["QV Notif 2-Excp"]       = "1"
+    out["QV Notif 2-Delv"]       = "1"
+
+    # All other columns remain blank by default (already created)
+
+    # Export to a temp file with NO HEADER
+    tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".csv")
+    out.to_csv(tmp.name, index=False, header=False, encoding="utf-8-sig")
+    tmp.close()
+    return tmp.name
+
+# ---------- GRADIO UI ----------
+TITLE = "UPS Batch CSV Converter (Import-ready, No Header)"
+DESC = (
+    "Upload your shipment CSV. The app will clean and convert it to UPS Batch format "
+    "(**exact column order** and **no header**), including: ZIP padding, address wrap ≤35 chars, "
+    "removing stray characters (e.g. ÿ), converting Dry Ice Weight (lbs→kg, rounded), "
+    "Service=01, Packaging Type=2, Delivery Confirm=S, QV Notif flags=1, QV Notif 1-Addr from Email, "
+    "QV Notif 2-Addr fixed to shaqdong@apexglobe.com, ADL Language blank."
+)
 
 demo = gr.Interface(
-    fn=clean_csv,
-    inputs=gr.File(label="📤 Upload CSV File"),
-    outputs=gr.File(label="📥 Download Cleaned CSV"),
-    title=title,
-    description=description,
+    fn=build_ups_batch_no_header,
+    inputs=gr.File(label="📤 Upload Source CSV"),
+    outputs=gr.File(label="📥 Download UPS Import-Ready CSV (No Header)"),
+    title=TITLE,
+    description=DESC,
     allow_flagging="never"
 )