Spaces:
Sleeping
Sleeping
File size: 4,815 Bytes
c719792 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 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) |