app.py

import gradio as gr
import pandas as pd
import re
import unicodedata
import tempfile

# ---------- 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 'ÿ', control chars and normalize to printable ASCII."""
    if pd.isna(s):
        return ""
    s = str(s).replace("ÿ", "")
    s = unicodedata.normalize("NFKD", s)
    s = "".join(ch for ch in s if 32 <= ord(ch) <= 126)
    return s.strip()

def format_zip(zip_code) -> str:
    """Pad to 5 digits; strip non-digits first."""
    if pd.isna(zip_code):
        return ""
    z = re.sub(r"[^\d]", "", str(zip_code).strip())
    if not z:
        return ""
    return z.zfill(5)[:5]
    
# Validate email format (must contain '@' and end with .domain)
def validate_email(s):
    s = clean_text(s)
    if not s or not re.match(r"^[^@]+@[^@]+\.[a-zA-Z]{2,}$", s):
        return ""  # blank out invalid emails
    return s
    
def flow_address_lines(lines, maxlen=35, maxlines=3):
    """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 = ["", "", ""]
    i = 0
    for tok in tokens:
        while len(tok) > maxlen:
            chunk, tok = tok[:maxlen], tok[maxlen:]
            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:
            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:
        df = pd.read_csv(file.name, encoding="utf-8-sig")

    df.columns = df.columns.str.strip()

    # 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","")
        ])
        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").apply(validate_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")

    # Set QV Notif 1 flags conditionally
    out["QV Notif 1-Ship"] = out["QV Notif 1-Addr"].apply(lambda x: "1" if x else "")
    out["QV Notif 1-Excp"] = out["QV Notif 1-Addr"].apply(lambda x: "1" if x else "")
    out["QV Notif 1-Delv"] = out["QV Notif 1-Addr"].apply(lambda x: "1" if x else "")

    out["QV Notif 2-Addr"]       = "wh-ord@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=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"
)

if __name__ == "__main__":
    demo.launch()