Spaces:
Sleeping
Sleeping
| 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() | |