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