Create utils.py

public_api/utils.py

@@ -0,0 +1,154 @@
+# utils.py
+import arrow
+def format_bot_response(text, source="SYSTEM", data=None, state=None, debug="", followup="",show_export=False,type=""):
+    """Ensures every return has the exact same structure."""
+    return {
+        "bot_response": text,
+        "source": source,
+        "data": data,
+        "state": state,
+        "debug": debug,
+        "followup": followup,
+        "show_export": show_export,
+        "type":type
+    }
+
+def flatten_json(y, prefix=''):
+    """Flatten nested JSON into dot-separated keys."""
+    out = {}
+    if isinstance(y, dict):
+        for k, v in y.items():
+            out.update(flatten_json(v, f"{prefix}{k}." if prefix else k + "."))
+    elif isinstance(y, list):
+        for i, v in enumerate(y):
+            out.update(flatten_json(v, f"{prefix}{i}." if prefix else f"{i}."))
+    else:
+        out[prefix[:-1]] = y  # remove trailing dot
+    return out
+
+def extract_fields(flat_data, mapping, formatters=None, requested_order=None):
+    result = {}
+    if formatters is None:
+        formatters = {}
+
+    # Determine the order: use requested_order if provided, otherwise the mapping order
+    display_keys = requested_order if requested_order else mapping.keys()
+
+    for display_name in display_keys:
+    # Get the JSON path from your mapping for this specific display name
+        json_path = mapping.get(display_name)
+        if not json_path:
+            continue
+
+        # Get the raw value (could be None if the key doesn't exist in the JSON)
+        val = flat_data.get(json_path)
+
+        # Apply Human Translation (humanize_value handles the None -> "N/A" logic)
+        if display_name in formatters:
+            try:
+                val = formatters[display_name](val)
+            except Exception:
+                # Keep original value if formatter crashes
+                pass
+
+        result[display_name] = val
+        
+    return result
+
+def humanize_value(val):
+    """
+    Handles Booleans, messy strings (NOT_REQUIRED), and preserves 0.
+    """
+    # Handle Strict Booleans (True/False)
+    if isinstance(val, bool):
+        return "Yes" if val else "No"
+    
+    # Handle specific "Useless" Strings
+    if isinstance(val, str) and val.upper() in ['NOT_REQUIRED', 'NONE', 'TIME_NONE', '']:
+        return "N/A"
+    
+    # Handle Numbers (preserve 0, but None becomes N/A)
+    if val is None:
+        return "N/A"
+
+    return val
+
+# Helper for Velocity Codes (The "B" issue)
+def translate_velocity(code):
+    velocity_map = {
+        'A': 'Fast Moving',
+        'B': 'Medium Moving',
+        'C': 'Slow Moving'
+    }
+    # Return the mapped text, or default to the original code if not found
+    return velocity_map.get(code, code)
+
+"""
+def extract_fields(flat_data: dict, user_fields: list, mapping: dict):
+    #Return only requested fields based on user-friendly mapping.
+    result = {}
+    for field in user_fields:
+        key = mapping.get(field.lower())
+        if key:
+            result[field] = flat_data.get(key, "Not available")
+        else:
+            result[field] = "Field not recognized"
+    return result
+"""
+
+"""
+def extract_fields(flat_data, mapping, formatters=None):
+    result = {}
+    if formatters is None:
+        formatters = {}
+
+    for display_name, json_path in mapping.items():
+        val = flat_data.get(json_path)
+
+        # 2. Check if data exists
+        if val is None:
+            result[display_name] = "Not available" 
+            continue
+
+        # 3. Apply Human Translation (if defined)
+        if display_name in formatters:
+            try:
+                # Call the specific formatter function for this field
+                val = formatters[display_name](val)
+            except Exception as e:
+                # If formatting fails, just keep original value to be safe
+                pass
+
+        result[display_name] = val
+        
+    return result
+"""
+
+def format_hold_type(val):
+    mapping = {"A":"Allocation Hold","C":"Critical Hold","D":"Customs Hold","M":"Non Movable","O":"RF Outbound Audit Hold","Q": "QA Hold","S":"Shipping Hold","SC":"Non status changeble"}
+
+    return mapping.get(val, val) # Returns "QA Hold" if Q, else original value
+
+def format_reason_code(val):
+    mapping = {"QUAL": "Quality Hold", "OUT-AUD-HLD": "RF Outbount Audit Hold","PRDLIN-HOLD":"Production Line Hold"}
+    return mapping.get(val, val)
+
+
+
+def format_backend_date(date):
+    # Arrow automatically recognizes the ISO format from your DB/API
+    date_obj = arrow.get(date)
+    
+    # Format to exactly: DD-MM-YYYY HH:MM:SS AM/PM
+    return date_obj.format('DD-MM-YYYY hh:mm:ssA')
+
+def translate_loc_sts(val):
+    mapping={
+        "E":"Empty",
+        "F":"Full",
+        "I":"Inventory Error",
+        "L":"Locked",
+        "N":"Pending",
+        "P":"Partially Full",
+    }
+    return mapping.get(val, val)