import base64
import pandas as pd
from pydantic import ValidationError
def process_spreadsheet_with_legacy_safeguards(file_path_or_buffer):
"""
Imports .xls or CSV data dumps from legacy systems, captures strict length
metrics for anomaly detection, and handles binary BLOB substitutions.
"""
# Read spreadsheet explicitly handling encoding where applicable
df = pd.read_excel(file_path_or_buffer)
records = df.to_dict(orient="records")
normalized_records = []
for row in records:
clean_row = {}
for k, v in row.items():
normalized_key = str(k).strip()
if pd.isna(v):
clean_row[normalized_key] = None
elif isinstance(v, bytes):
# If binary data is passed, convert to Base64 string for safe JSON transport
clean_row[normalized_key] = base64.b64encode(v).decode('utf-8')
elif isinstance(v, str):
# Ensure proper UTF-8 handling and strip trailing EBCDIC/ASCII padding artifacts
clean_row[normalized_key] = v.strip()
else:
# Handle numeric coercions (e.g., spreadsheet floats like 1048576.0 -> int)
clean_row[normalized_key] = v
normalized_records.append(clean_row)
return normalized_records
def validate_and_capture_lengths(records: list):
"""
Validates records against the DTO and logs exact field lengths
instead of just item counts to catch truncation and packing anomalies.
"""
anomalies = []
for index, record in enumerate(records):
try:
CustomerResponseSchema.model_validate(record)
except ValidationError as err:
# Capture detailed metadata including exact length of every field
field_length_metrics = {}
for k, v in record.items():
if isinstance(v, str):
field_length_metrics[k] = {"length": len(v), "preview": v[:20]}
elif v is None:
field_length_metrics[k] = {"length": 0, "value": "null"}
else:
field_length_metrics[k] = {"length": len(str(v)), "value": v}
anomalies.append({
"record_index": index,
"field_metrics": field_length_metrics, # Replaces simple item counts with actual lengths
"validation_errors": err.errors(),
})
return anomalies