"""Formatting, validation and type-conversion helpers for the export pipeline."""
import json
import pandas as pd
from pathlib import Path
from typing import Optional
[docs]
def validate_doc_id(doc_id: str) -> bool:
"""Validate a BusinessObjects document identifier.
:param doc_id: Identifier to validate
:type doc_id: str
:return: True if the identifier is valid
:rtype: bool
Example::
>>> validate_doc_id("123456")
True
>>> validate_doc_id("")
False
"""
if not doc_id or not isinstance(doc_id, str):
return False
# Must be alphanumeric (may contain _ or -)
return doc_id.replace('_', '').replace('-', '').isalnum()
[docs]
def parse_bo_error(response_data: dict) -> str:
"""Extract and format an error message from the API.
:param response_data: JSON data of the error response
:type response_data: dict
:return: Formatted error message
:rtype: str
Example::
>>> error = {"error_code": "ERR_WIS_30270", "message": "Invalid session"}
>>> parse_bo_error(error)
"BO error ERR_WIS_30270: Invalid session"
"""
error_code = response_data.get('error_code', 'N/A')
message = response_data.get('message', 'Unknown error')
details = response_data.get('details', '')
error_msg = f"BO error {error_code}: {message}"
if details:
error_msg += f"\nDetails: {details}"
return error_msg
[docs]
def truncate_string(text: str, max_length: int = 100, suffix: str = "...") -> str:
"""Truncate a string if it exceeds the maximum length.
:param text: Text to truncate
:type text: str
:param max_length: Maximum length including the suffix
:type max_length: int
:param suffix: Suffix appended on truncation
:type suffix: str
:return: Truncated text if needed
:rtype: str
Example::
>>> truncate_string("A very long text", max_length=10)
"A very ..."
"""
if len(text) <= max_length:
return text
return text[:max_length - len(suffix)] + suffix
[docs]
def load_column_mapping(mapping_file=None) -> dict:
"""Load the column mapping configuration.
Accepts a path to a JSON file or an already-built mapping dictionary.
:param mapping_file: Path to the JSON file, a mapping dict, or None to use
column_mapping.json
:return: Mapping configuration dictionary
:rtype: dict
Example::
>>> mapping = load_column_mapping()
>>> mapping['Code INSEE']['normalized_name']
'insee_code'
"""
if isinstance(mapping_file, dict):
return mapping_file
if mapping_file is None:
# Default path: config/column_mapping.json
project_root = Path(__file__).parent.parent
mapping_file = project_root / "config" / "column_mapping.json"
mapping_path = Path(mapping_file)
if not mapping_path.exists():
return {}
with open(mapping_path, 'r', encoding='utf-8') as f:
return json.load(f)
[docs]
def get_dtype_dict(mapping_file: Optional[str] = None) -> dict:
"""Build a dtype dictionary for pd.read_csv().
Forces columns that must preserve their format (codes with leading zeros,
categories, booleans, dates, integers) to load as str. Floats are left to
pandas to handle decimal=',' and thousands=' '.
The final conversion to the target type is done by apply_column_mapping().
:param mapping_file: Path to the mapping file, optional
:type mapping_file: Optional[str]
:return: Mapping {original_column_name: pandas_dtype}
:rtype: dict
Example::
>>> dtype_dict = get_dtype_dict()
>>> dtype_dict['Code Postal']
str
>>> dtype_dict['Code INSEE']
str
"""
mapping = load_column_mapping(mapping_file)
if not mapping:
return {}
dtype_dict = {}
for original_name, config in mapping.items():
python_type = config.get('python_type')
# Force str for every type except float to preserve the format:
# - str: avoid automatic numeric conversion (postal codes, INSEE, etc.)
# - int: preserve leading zeros (department "01", etc.)
# - category: avoid incorrect inference
# - bool: keep text values ("O"/"N") for later conversion
# - datetime: keep as text for later parsing with a specific format
# - float: let pandas handle decimal=',' and thousands=' '
if python_type in ['str', 'int', 'category', 'bool', 'datetime']:
dtype_dict[original_name] = str
return dtype_dict
[docs]
def apply_column_mapping(df: pd.DataFrame, mapping_file=None, date_format: Optional[str] = None) -> pd.DataFrame:
"""Apply the column mapping and type conversions.
Renames columns according to normalized_name and converts types according
to python_type. A per-column date_format overrides the global one.
Columns absent from the mapping are left unchanged.
:param df: DataFrame to transform
:type df: pd.DataFrame
:param mapping_file: Path to the mapping file, a mapping dict, or None
:param date_format: Default date format for datetime columns, optional
:type date_format: Optional[str]
:return: DataFrame with renamed and typed columns
:rtype: pd.DataFrame
Example::
>>> df = pd.DataFrame({'Code INSEE': ['01234', '56789']})
>>> df_mapped = apply_column_mapping(df)
>>> 'insee_code' in df_mapped.columns
True
"""
mapping = load_column_mapping(mapping_file)
if not mapping:
return df
df = df.copy()
rename_dict = {}
column_config = {}
for original_name, config in mapping.items():
if original_name in df.columns:
new_name = config.get('normalized_name', original_name)
rename_dict[original_name] = new_name
column_config[new_name] = config
df = df.rename(columns=rename_dict)
for col_name, config in column_config.items():
if col_name not in df.columns:
continue
python_type = config.get('python_type')
try:
if python_type == 'category':
df[col_name] = df[col_name].astype('category')
elif python_type == 'str':
df[col_name] = df[col_name].astype(str)
elif python_type == 'int':
# Replace commas with dots on non-numeric columns
if not pd.api.types.is_numeric_dtype(df[col_name]):
df[col_name] = df[col_name].astype(str).str.replace(',', '.', regex=False)
# Convert to float first, then round and convert to Int64;
# this handles cases where to_numeric returns float64 with .0
numeric_col = pd.to_numeric(df[col_name], errors='coerce')
# Round to avoid floating-point precision issues
numeric_col = numeric_col.round(0)
df[col_name] = numeric_col.astype('Int64')
elif python_type == 'float':
# Replace commas with dots on non-numeric columns
if not pd.api.types.is_numeric_dtype(df[col_name]):
df[col_name] = df[col_name].astype(str).str.replace(',', '.', regex=False)
df[col_name] = pd.to_numeric(df[col_name], errors='coerce')
elif python_type == 'bool':
df[col_name] = df[col_name].map({'O': True, 'N': False})
elif python_type == 'datetime':
if 'hour' in col_name or 'heure' in col_name.lower():
pass
else:
fmt = config.get('date_format') or date_format or '%m/%d/%y'
df[col_name] = pd.to_datetime(df[col_name], format=fmt)
except Exception as e:
print(f"Warning: could not convert {col_name} to {python_type}: {e}")
return df