"""Export of BusinessObjects data to Parquet, CSV or JSON, with chunked processing for large volumes."""
import io
import os
import tempfile
import zipfile
import pandas as pd
from pathlib import Path
from typing import TYPE_CHECKING, Optional, List
from .utils import apply_column_mapping, get_dtype_dict
if TYPE_CHECKING:
from .client import BOAPIClient
def _ensure_parent_dir(path: str) -> None:
"""Create the parent directory of a file path if it does not exist.
:param path: Output file path
:type path: str
"""
parent = os.path.dirname(path)
if parent:
os.makedirs(parent, exist_ok=True)
[docs]
class ExportManager:
"""Manager for data exports.
Exports data to a pandas DataFrame, CSV, Parquet or JSON with column
mapping and automatic typing.
"""
def __init__(self, client: 'BOAPIClient'):
"""Initialize the manager.
:param client: BO client instance
:type client: BOAPIClient
"""
self.client = client
[docs]
def get_csv_from_zip(self, zip_content: bytes) -> str:
"""Extract the CSV content from an in-memory ZIP archive.
:param zip_content: Binary content of the ZIP file
:type zip_content: bytes
:return: CSV file content
:rtype: str
:raises Exception: If no CSV file is found
"""
with zipfile.ZipFile(io.BytesIO(zip_content)) as zip_ref:
file_list = zip_ref.namelist()
csv_files = [f for f in file_list if f.endswith('.csv')]
if not csv_files:
raise Exception(f"No CSV file found in the ZIP. Files present: {file_list}")
# Read the first CSV found
csv_filename = csv_files[0]
csv_content = zip_ref.read(csv_filename).decode('utf-8')
return csv_content
[docs]
def get_raw_data(self, doc_id: str, use_cache: bool = True) -> str:
"""Retrieve the raw data as CSV.
Downloads the data from the API and keeps it in memory. Automatically
extracts from a ZIP archive when needed.
:param doc_id: Document identifier
:type doc_id: str
:param use_cache: Enable cache usage
:type use_cache: bool
:return: CSV content
:rtype: str
:raises Exception: If the download fails
"""
if use_cache and self.client.cache.has(doc_id):
print("Data retrieved from cache")
return self.client.cache.get(doc_id)
print("Downloading data from BO...")
url = f"{self.client.base_url}/raylight/v1/documents/{doc_id}"
response = self.client.session.get(
url,
headers=self.client.get_headers(accept="text/csv", use_quotes=False)
)
if response.status_code != 200:
raise Exception(f"Export error: {response.status_code} - {response.text}")
content_type = response.headers.get('Content-Type', '')
if 'zip' in content_type.lower() or content_type == 'application/octet-stream':
print("Extracting CSV from the ZIP (in memory)...")
csv_content = self.get_csv_from_zip(response.content)
elif 'text' in content_type or 'csv' in content_type:
print("Direct CSV")
csv_content = response.text
else:
raise Exception(f"Unsupported content type: {content_type}")
if use_cache:
self.client.cache.set(doc_id, csv_content)
print(f"Data cached ({len(csv_content):,} characters)")
return csv_content
[docs]
def get_dataframe(self, doc_id: str, separator: str = None,
use_cache: bool = True, debug: bool = False,
apply_mapping: bool = True, mapping_file: Optional[str] = None,
date_format: Optional[str] = None) -> pd.DataFrame:
"""Retrieve the data and convert it to a DataFrame.
Downloads the data, auto-detects the CSV separator and applies the
column mapping with typing.
:param doc_id: Document identifier
:type doc_id: str
:param separator: CSV separator, auto-detected if None
:type separator: str
:param use_cache: Enable cache usage
:type use_cache: bool
:param debug: Enable debug output
:type debug: bool
:param apply_mapping: Enable column mapping and typing
:type apply_mapping: bool
:param mapping_file: Path to the mapping file, optional
:type mapping_file: Optional[str]
:param date_format: Date format used for datetime columns, optional
:type date_format: Optional[str]
:return: DataFrame with the data
:rtype: pd.DataFrame
:raises Exception: If CSV parsing fails
"""
csv_content = self.get_raw_data(doc_id, use_cache)
print("Converting to DataFrame...")
dtype_dict = get_dtype_dict(mapping_file) if apply_mapping else {}
# Separators to try (semicolon first for the French format)
separators = [separator] if separator else [';', ',', '\t', '|']
for sep in separators:
try:
df = pd.read_csv(
io.StringIO(csv_content),
sep=sep,
on_bad_lines='skip',
encoding='utf-8',
decimal=',', # French format: comma as decimal separator
thousands=' ', # French format: space as thousands separator
low_memory=False, # Avoid mixed-type warnings
dtype=dtype_dict # Force str types at load time
)
if df.shape[1] > 1 or (df.shape[1] == 1 and len(df) > 0):
print(f"DataFrame created: {df.shape} (separator='{sep}')")
if apply_mapping:
print("Applying column mapping...")
df = apply_column_mapping(df, mapping_file, date_format)
print("Columns renamed and typed")
if debug:
print(f"\nColumns: {df.columns.tolist()}")
print("\nPreview:")
print(df.head())
return df
except Exception as e:
if separator:
# A specific separator was requested: raise the error
raise Exception(f"Error with separator '{sep}': {e}")
# Otherwise try the next one
continue
raise Exception("Could not parse the CSV with the tested separators")
[docs]
def save_to_parquet(self, doc_id: str, output_path: str,
use_cache: bool = True, apply_mapping: bool = True,
mapping_file: Optional[str] = None, chunk_size: Optional[int] = None,
date_format: Optional[str] = None,
**kwargs) -> str:
"""Save the data to Parquet format.
Converts and saves the data with compression. If chunk_size is given,
processes the data in batches to optimize memory usage.
:param doc_id: Document identifier
:type doc_id: str
:param output_path: Output file path
:type output_path: str
:param use_cache: Enable cache usage
:type use_cache: bool
:param apply_mapping: Enable column mapping
:type apply_mapping: bool
:param mapping_file: Path to the mapping file, optional
:type mapping_file: Optional[str]
:param chunk_size: Chunk size for batch processing
:type chunk_size: Optional[int]
:param date_format: Date format used for datetime columns, optional
:type date_format: Optional[str]
:param kwargs: Extra arguments for to_parquet
:return: Path of the created file
:rtype: str
"""
if chunk_size:
return self._save_to_parquet_chunked(
doc_id, output_path, use_cache, apply_mapping, mapping_file, chunk_size, date_format, **kwargs
)
print("Parquet export...")
df = self.get_dataframe(doc_id, use_cache=use_cache, apply_mapping=apply_mapping, mapping_file=mapping_file, date_format=date_format)
parquet_path = output_path if output_path.endswith('.parquet') else f"{output_path}.parquet"
parquet_kwargs = {
'compression': 'snappy', # Good compression, fast
'index': False,
'engine': 'pyarrow'
}
parquet_kwargs.update(kwargs)
# Convert object columns to string while preserving NULL values;
# NaN/None must become NULL in PostgreSQL, not the string "nan"
for col in df.select_dtypes(include=['object']).columns:
df[col] = df[col].apply(lambda x: str(x) if pd.notna(x) else None)
# Replace "XXXXXX" placeholders (missing data) with None
for col in df.select_dtypes(include=['object']).columns:
df[col] = df[col].apply(lambda x: None if x == "XXXXXX" else x)
# Convert every NaN to None, including numeric columns
df = df.where(pd.notna(df), None)
_ensure_parent_dir(parquet_path)
df.to_parquet(parquet_path, **parquet_kwargs)
size_mb = os.path.getsize(parquet_path) / 1024 / 1024
print(f"File created: {parquet_path} ({size_mb:.2f} MB)")
return parquet_path
def _save_to_parquet_chunked(self, doc_id: str, output_path: str,
use_cache: bool, apply_mapping: bool,
mapping_file: Optional[str], chunk_size: int,
date_format: Optional[str],
**kwargs) -> str:
"""Save the data in chunks to optimize memory usage.
Internal method that processes the data in batches to handle large
volumes without exhausting memory.
:param doc_id: Document identifier
:type doc_id: str
:param output_path: Output file path
:type output_path: str
:param use_cache: Enable cache usage
:type use_cache: bool
:param apply_mapping: Enable column mapping
:type apply_mapping: bool
:param mapping_file: Path to the mapping file
:type mapping_file: Optional[str]
:param chunk_size: Number of rows per chunk
:type chunk_size: int
:param date_format: Date format used for datetime columns, optional
:type date_format: Optional[str]
:param kwargs: Extra arguments for to_parquet
:return: Path of the created file
:rtype: str
"""
print(f"Chunked Parquet export (chunk size: {chunk_size:,} rows)...")
csv_content = self.get_raw_data(doc_id, use_cache)
parquet_path = output_path if output_path.endswith('.parquet') else f"{output_path}.parquet"
dtype_dict = get_dtype_dict(mapping_file) if apply_mapping else {}
temp_dir = tempfile.mkdtemp(prefix="bo_chunks_")
temp_chunks: List[str] = []
try:
# Detect the separator by reading the first rows
separators = [';', ',', '\t', '|']
detected_sep = None
for sep in separators:
try:
test_df = pd.read_csv(
io.StringIO(csv_content),
sep=sep,
nrows=5,
on_bad_lines='skip',
encoding='utf-8',
dtype=dtype_dict # Force str types even for the test
)
if test_df.shape[1] > 1:
detected_sep = sep
break
except Exception:
continue
if not detected_sep:
raise Exception("Could not detect the CSV separator")
print(f"Separator detected: '{detected_sep}'")
print("Reading and processing in chunks...")
chunk_reader = pd.read_csv(
io.StringIO(csv_content),
sep=detected_sep,
chunksize=chunk_size,
on_bad_lines='skip',
encoding='utf-8',
decimal=',',
thousands=' ',
low_memory=False,
dtype=dtype_dict # Force str types at load time
)
chunk_num = 0
total_rows = 0
for chunk_df in chunk_reader:
if apply_mapping:
chunk_df = apply_column_mapping(chunk_df, mapping_file, date_format)
# Convert object columns to string while preserving NULL values;
# NaN/None must become NULL in PostgreSQL, not the string "nan"
for col in chunk_df.select_dtypes(include=['object']).columns:
chunk_df[col] = chunk_df[col].apply(lambda x: str(x) if pd.notna(x) else None)
# Replace "XXXXXX" placeholders (missing data) with None
for col in chunk_df.select_dtypes(include=['object']).columns:
chunk_df[col] = chunk_df[col].apply(lambda x: None if x == "XXXXXX" else x)
# Convert every NaN to None, including numeric columns
chunk_df = chunk_df.where(pd.notna(chunk_df), None)
chunk_path = os.path.join(temp_dir, f"chunk_{chunk_num:04d}.parquet")
chunk_df.to_parquet(
chunk_path,
compression='snappy',
index=False,
engine='pyarrow'
)
temp_chunks.append(chunk_path)
total_rows += len(chunk_df)
chunk_num += 1
print(f" Chunk {chunk_num}: {len(chunk_df):,} rows processed (total: {total_rows:,})")
print(f"{chunk_num} chunks created ({total_rows:,} rows in total)")
print("Aggregating chunks...")
all_chunks = [pd.read_parquet(chunk_path) for chunk_path in temp_chunks]
final_df = pd.concat(all_chunks, ignore_index=True)
# Re-apply the mapping after concatenation to enforce consistent types.
# This is needed because pandas may convert to 'object' during concat
# when chunks have slightly different types (int64 vs float64).
if apply_mapping:
print("Re-applying type mapping after concatenation...")
final_df = apply_column_mapping(final_df, mapping_file, date_format)
parquet_kwargs = {
'compression': 'snappy',
'index': False,
'engine': 'pyarrow'
}
parquet_kwargs.update(kwargs)
_ensure_parent_dir(parquet_path)
final_df.to_parquet(parquet_path, **parquet_kwargs)
size_mb = os.path.getsize(parquet_path) / 1024 / 1024
print(f"Final file created: {parquet_path} ({size_mb:.2f} MB)")
return parquet_path
finally:
print("Cleaning up temporary files...")
for chunk_path in temp_chunks:
try:
os.remove(chunk_path)
except Exception:
pass
try:
os.rmdir(temp_dir)
except Exception:
pass
print("Cleanup done")
[docs]
def save_to_csv(self, doc_id: str, output_path: str, use_cache: bool = True) -> str:
"""Save the data to CSV format.
:param doc_id: Document identifier
:type doc_id: str
:param output_path: Output file path
:type output_path: str
:param use_cache: Enable cache usage
:type use_cache: bool
:return: Path of the created file
:rtype: str
"""
print("CSV export...")
# The raw data is already CSV
csv_content = self.get_raw_data(doc_id, use_cache)
csv_path = output_path if output_path.endswith('.csv') else f"{output_path}.csv"
_ensure_parent_dir(csv_path)
with open(csv_path, 'w', encoding='utf-8') as f:
f.write(csv_content)
size_mb = os.path.getsize(csv_path) / 1024 / 1024
print(f"File created: {csv_path} ({size_mb:.2f} MB)")
return csv_path
[docs]
def save_to_json(self, doc_id: str, output_path: str,
use_cache: bool = True, apply_mapping: bool = True,
mapping_file: Optional[str] = None, date_format: Optional[str] = None,
orient: str = 'records') -> str:
"""Save the data to JSON format.
:param doc_id: Document identifier
:type doc_id: str
:param output_path: Output file path
:type output_path: str
:param use_cache: Enable cache usage
:type use_cache: bool
:param apply_mapping: Enable column mapping
:type apply_mapping: bool
:param mapping_file: Path to the mapping file, optional
:type mapping_file: Optional[str]
:param date_format: Date format used for datetime columns, optional
:type date_format: Optional[str]
:param orient: JSON format (records, split, index, columns, values, table)
:type orient: str
:return: Path of the created file
:rtype: str
"""
print("JSON export...")
df = self.get_dataframe(doc_id, use_cache=use_cache, apply_mapping=apply_mapping, mapping_file=mapping_file, date_format=date_format)
json_path = output_path if output_path.endswith('.json') else f"{output_path}.json"
_ensure_parent_dir(json_path)
df.to_json(json_path, orient=orient, force_ascii=False, indent=2)
size_mb = os.path.getsize(json_path) / 1024 / 1024
print(f"File created: {json_path} ({size_mb:.2f} MB)")
return json_path