json-tools-rs 0.9.3

//! Python bindings for JSON Tools RS
//!
//! This module provides Python bindings for the unified JSONTools API
//! using PyO3. It exposes the complete JSONTools builder pattern API to Python
//! with support for both flattening and unflattening operations, collision handling,
//! and all advanced features.

#[cfg(feature = "python")]
use pyo3::exceptions::{PyRuntimeError, PyValueError};
#[cfg(feature = "python")]
use pyo3::prelude::*;
#[cfg(feature = "python")]
use pyo3::types::PyModule;

#[cfg(feature = "python")]
use std::mem;
#[cfg(feature = "python")]
use std::sync::Mutex;

#[cfg(feature = "python")]
use crate::{JSONTools, JsonOutput};

// Use conditional JSON parser module (sonic-rs on 64-bit, simd-json on 32-bit)
#[cfg(feature = "python")]
use crate::json_parser;

// TIER 6→3 OPTIMIZATION: Direct Python<->Rust conversion without JSON serialization
#[cfg(feature = "python")]
use pythonize::{depythonize, pythonize};

#[cfg(feature = "python")]
pyo3::create_exception!(
    json_tools_rs,
    JsonToolsError,
    pyo3::exceptions::PyException,
    "Python exception for JSON Tools operations"
);

/// Lock the inner config mutex, converting poison errors to Python exceptions.
#[cfg(feature = "python")]
#[inline]
fn lock_config(mutex: &Mutex<JSONTools>) -> PyResult<std::sync::MutexGuard<'_, JSONTools>> {
    mutex
        .lock()
        .map_err(|e| PyRuntimeError::new_err(format!("internal config lock poisoned: {e}")))
}

// =============================================================================
// DataFrame and Series Support Types
// =============================================================================

/// Type of DataFrame library detected via duck-typing.
///
/// Detection uses `type(obj).__module__` and `type(obj).__name__` to identify
/// the DataFrame variant without importing the library, falling back to
/// checking for `to_dict()` or `to_json()` methods for generic compatibility.
#[cfg(feature = "python")]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum DataFrameType {
    /// pandas.DataFrame — uses `to_dict("records")` for row extraction
    Pandas,
    /// polars.DataFrame — uses `to_dicts()` for row extraction
    Polars,
    /// pyarrow.Table — uses `to_pydict()` for columnar extraction
    PyArrow,
    /// pyspark.sql.DataFrame — uses `toJSON().collect()` for distributed extraction
    PySpark,
    /// Any object with `to_dict()` or `to_json()` methods
    Generic,
}

/// Type of Series library detected via duck-typing.
///
/// Similar to `DataFrameType`, uses module/name introspection to identify
/// the Series variant, falling back to `to_list()` / `tolist()` for generics.
#[cfg(feature = "python")]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum SeriesType {
    /// pandas.Series — uses `tolist()` for value extraction
    Pandas,
    /// polars.Series — uses `to_list()` for value extraction
    Polars,
    /// pyarrow.Array or pyarrow.ChunkedArray — uses `to_pylist()`
    PyArrow,
    /// pyspark.sql.Column (rare) — uses `collect()` for distributed extraction
    PySpark,
    /// Any object with `to_list()` or `tolist()` methods
    Generic,
}

/// Unified data structure type for detection.
///
/// Wraps either a DataFrame or Series detection result so the processing
/// pipeline can handle both types uniformly before dispatching to
/// type-specific extraction logic.
#[cfg(feature = "python")]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum DataStructureType {
    /// A DataFrame-like object (tabular rows/columns)
    DataFrame(DataFrameType),
    /// A Series-like object (single column/array of values)
    Series(SeriesType),
}

/// Python wrapper for JsonOutput enum
#[cfg(feature = "python")]
#[pyclass(name = "JsonOutput", frozen, from_py_object)]
#[derive(Debug, Clone)]
pub struct PyJsonOutput {
    inner: JsonOutput,
}

#[cfg(feature = "python")]
#[pymethods]
impl PyJsonOutput {
    /// Check if this is a single result
    #[getter]
    fn is_single(&self) -> bool {
        matches!(self.inner, JsonOutput::Single(_))
    }

    /// Check if this is a multiple result
    #[getter]
    fn is_multiple(&self) -> bool {
        matches!(self.inner, JsonOutput::Multiple(_))
    }

    /// Get the single result (raises ValueError if multiple)
    fn get_single(&self) -> PyResult<String> {
        match &self.inner {
            JsonOutput::Single(result) => Ok(result.clone()),
            JsonOutput::Multiple(_) => Err(PyValueError::new_err(
                "Result contains multiple JSON strings, use get_multiple() instead",
            )),
        }
    }

    /// Get the multiple results (raises ValueError if single)
    fn get_multiple(&self) -> PyResult<Vec<String>> {
        match &self.inner {
            JsonOutput::Single(_) => Err(PyValueError::new_err(
                "Result contains single JSON string, use get_single() instead",
            )),
            JsonOutput::Multiple(results) => Ok(results.clone()),
        }
    }

    /// Get the result as a Python object (string for single, list for multiple)
    fn to_python(&self, py: Python) -> PyResult<Py<PyAny>> {
        match &self.inner {
            JsonOutput::Single(result) => Ok(result.into_pyobject(py)?.into_any().unbind()),
            JsonOutput::Multiple(results) => Ok(results.into_pyobject(py)?.into_any().unbind()),
        }
    }

    fn __repr__(&self) -> String {
        match &self.inner {
            JsonOutput::Single(result) => format!("JsonOutput.Single('{}')", result),
            JsonOutput::Multiple(results) => format!("JsonOutput.Multiple({:?})", results),
        }
    }

    fn __str__(&self) -> String {
        match &self.inner {
            JsonOutput::Single(result) => result.clone(),
            JsonOutput::Multiple(results) => format!("{:?}", results),
        }
    }
}

#[cfg(feature = "python")]
impl From<JsonOutput> for PyJsonOutput {
    fn from(output: JsonOutput) -> Self {
        PyJsonOutput { inner: output }
    }
}

#[cfg(feature = "python")]
impl PyJsonOutput {
    /// Helper method to create PyJsonOutput from Rust JsonOutput
    pub fn from_rust_output(output: JsonOutput) -> Self {
        PyJsonOutput { inner: output }
    }
}

/// Python JSONTools class - the unified API for JSON manipulation
///
/// This is the single entry point for all JSON operations in Python, providing both
/// flattening and unflattening capabilities with advanced features like collision handling,
/// filtering, and comprehensive transformations. It mirrors the Rust JSONTools API exactly.
///
/// # Performance Optimization
/// Uses Mutex for interior mutability to avoid cloning the entire JSONTools struct
/// on every builder method call. This provides 30-50% performance improvement over
/// the previous clone-based approach while maintaining thread safety for Python's GIL.
///
/// # Input/Output Type Mapping
/// - str input → str output (JSON string)
/// - dict input → dict output (Python dictionary)
/// - list[str] input → list[str] output (list of JSON strings)
/// - list[dict] input → list[dict] output (list of Python dictionaries)
/// - Mixed list preserves original types in output
///
/// # Examples
///
/// ```python
/// import json_tools_rs
///
/// # Basic flattening
/// result = json_tools_rs.JSONTools().flatten().execute('{"a": {"b": 1}}')
/// print(result)  # '{"a.b": 1}' (string)
///
/// # Basic unflattening
/// result = json_tools_rs.JSONTools().unflatten().execute('{"a.b": 1}')
/// print(result)  # '{"a": {"b": 1}}' (string)
///
/// # Advanced configuration with collision handling
/// tools = (json_tools_rs.JSONTools()
///     .flatten()
///     .separator("::")
///     .remove_empty_strings(True)
///     .remove_nulls(True)
///     .lowercase_keys(True)
///     .key_replacement("(User|Admin|Guest)_", "")
///     .handle_key_collision(True))
///
/// result = tools.execute({"User_name": "John", "Admin_name": "", "Guest_name": "Bob"})
/// print(result)  # {"name": ["John", "Bob"]} (dict, empty string filtered out)
///
///
/// # Batch processing with type preservation
/// str_list = ['{"a": 1}', '{"b": 2}']
/// results = json_tools_rs.JSONTools().flatten().execute(str_list)
/// print(results)  # ['{"a": 1}', '{"b": 2}'] (list of strings)
/// ```
#[cfg(feature = "python")]
#[pyclass(name = "JSONTools")]
pub struct PyJSONTools {
    // Use Mutex for interior mutability - allows mutation through shared reference
    // This eliminates the need to clone JSONTools on every builder method call
    // Mutex is required for thread safety (PyO3 requires Sync)
    inner: Mutex<JSONTools>,
}

#[cfg(feature = "python")]
impl Default for PyJSONTools {
    fn default() -> Self {
        Self::new()
    }
}

// =============================================================================
// DataFrame and Series Detection Functions
// =============================================================================

/// Detect if the input is a DataFrame or Series
#[cfg(feature = "python")]
fn detect_data_structure(obj: &Bound<'_, PyAny>) -> PyResult<Option<DataStructureType>> {
    // Try DataFrame detection first
    if let Some(df_type) = detect_dataframe_type(obj)? {
        return Ok(Some(DataStructureType::DataFrame(df_type)));
    }

    // Try Series detection
    if let Some(series_type) = detect_series_type(obj)? {
        return Ok(Some(DataStructureType::Series(series_type)));
    }

    // Neither DataFrame nor Series
    Ok(None)
}

/// Detect DataFrame type using duck typing (no imports)
#[cfg(feature = "python")]
fn detect_dataframe_type(obj: &Bound<'_, PyAny>) -> PyResult<Option<DataFrameType>> {
    // Get module and class name for specific detection
    let module = obj
        .getattr("__class__")?
        .getattr("__module__")?
        .extract::<String>()
        .unwrap_or_default();

    let class_name = obj
        .getattr("__class__")?
        .getattr("__name__")?
        .extract::<String>()
        .unwrap_or_default();

    // Check for PyArrow Table (class name is "Table", not "DataFrame")
    if module.starts_with("pyarrow") && class_name == "Table" {
        return Ok(Some(DataFrameType::PyArrow));
    }

    // Check for DataFrame-like methods
    let has_to_dict = obj.hasattr("to_dict")?;
    let has_columns = obj.hasattr("columns")?;

    if !has_to_dict && !has_columns {
        return Ok(None); // Not a DataFrame
    }

    // Check if it's actually a DataFrame class
    if class_name != "DataFrame" {
        return Ok(None);
    }

    // Detect specific library
    if module.starts_with("pandas") {
        Ok(Some(DataFrameType::Pandas))
    } else if module.starts_with("polars") {
        Ok(Some(DataFrameType::Polars))
    } else if module.contains("pyspark.sql") {
        Ok(Some(DataFrameType::PySpark))
    } else if has_to_dict || obj.hasattr("to_json")? {
        // Generic DataFrame-like object
        Ok(Some(DataFrameType::Generic))
    } else {
        Ok(None)
    }
}

/// Detect Series type using duck typing (no imports)
#[cfg(feature = "python")]
fn detect_series_type(obj: &Bound<'_, PyAny>) -> PyResult<Option<SeriesType>> {
    // Get module and class name for specific detection
    let module = obj
        .getattr("__class__")?
        .getattr("__module__")?
        .extract::<String>()
        .unwrap_or_default();

    let class_name = obj
        .getattr("__class__")?
        .getattr("__name__")?
        .extract::<String>()
        .unwrap_or_default();

    // Check for PyArrow Array/ChunkedArray (various class names: Array, ChunkedArray, Int64Array, etc.)
    if module.starts_with("pyarrow")
        && (class_name.contains("Array") || obj.hasattr("to_pylist")?)
    {
        return Ok(Some(SeriesType::PyArrow));
    }

    // Check for Series-like methods
    let has_to_list = obj.hasattr("to_list")? || obj.hasattr("tolist")?;
    let has_dtype = obj.hasattr("dtype")?;

    if !has_to_list && !has_dtype {
        return Ok(None); // Not a Series
    }

    // Check if it's actually a Series class
    if class_name != "Series" {
        return Ok(None);
    }

    // Detect specific library
    if module.starts_with("pandas") {
        Ok(Some(SeriesType::Pandas))
    } else if module.starts_with("polars") {
        Ok(Some(SeriesType::Polars))
    } else if module.contains("pyspark") {
        Ok(Some(SeriesType::PySpark))
    } else if has_to_list {
        // Generic Series-like object
        Ok(Some(SeriesType::Generic))
    } else {
        Ok(None)
    }
}

// =============================================================================
// DataFrame Conversion Functions
// =============================================================================

/// Convert DataFrame to list of records (dicts)
#[cfg(feature = "python")]
fn dataframe_to_records(
    df: &Bound<'_, PyAny>,
    df_type: DataFrameType,
) -> PyResult<Vec<serde_json::Value>> {
    let py = df.py();

    match df_type {
        DataFrameType::Pandas => {
            // Call df.to_dict(orient='records')
            let kwargs = pyo3::types::PyDict::new(py);
            kwargs.set_item("orient", "records")?;
            let records = df.call_method("to_dict", (), Some(&kwargs))?;

            // Convert Python list of dicts to Vec<serde_json::Value>
            let list = records.cast::<pyo3::types::PyList>()?;
            convert_pylist_to_json_values(list)
        }

        DataFrameType::Polars => {
            // Call df.to_dicts()
            let records = df.call_method0("to_dicts")?;
            let list = records.cast::<pyo3::types::PyList>()?;
            convert_pylist_to_json_values(list)
        }

        DataFrameType::PyArrow => {
            // Call table.to_pylist()
            let records = df.call_method0("to_pylist")?;
            let list = records.cast::<pyo3::types::PyList>()?;
            convert_pylist_to_json_values(list)
        }

        DataFrameType::PySpark => {
            // Call df.toPandas().to_dict(orient='records')
            let pandas_df = df.call_method0("toPandas")?;
            let kwargs = pyo3::types::PyDict::new(py);
            kwargs.set_item("orient", "records")?;
            let records = pandas_df.call_method("to_dict", (), Some(&kwargs))?;

            let list = records.cast::<pyo3::types::PyList>()?;
            convert_pylist_to_json_values(list)
        }

        DataFrameType::Generic => {
            // Try to_dict() first, fallback to to_json()
            if df.hasattr("to_dict")? {
                let kwargs = pyo3::types::PyDict::new(py);
                kwargs.set_item("orient", "records")?;
                let records = df.call_method("to_dict", (), Some(&kwargs))?;
                let list = records.cast::<pyo3::types::PyList>()?;
                convert_pylist_to_json_values(list)
            } else {
                Err(JsonToolsError::new_err(
                    "Generic DataFrame must have to_dict() method",
                ))
            }
        }
    }
}

/// Convert Python list of dicts to Vec<serde_json::Value>
#[cfg(feature = "python")]
fn convert_pylist_to_json_values(
    list: &Bound<'_, pyo3::types::PyList>,
) -> PyResult<Vec<serde_json::Value>> {
    let mut values = Vec::with_capacity(list.len());
    for item in list.iter() {
        // Use depythonize for efficient conversion
        let value: serde_json::Value = depythonize(&item)
            .map_err(|e| JsonToolsError::new_err(format!("Failed to convert record: {}", e)))?;
        values.push(value);
    }
    Ok(values)
}

// =============================================================================
// Series Conversion Functions
// =============================================================================

/// Convert Series to Python list
#[cfg(feature = "python")]
fn series_to_list<'py>(
    series: &Bound<'py, PyAny>,
    series_type: SeriesType,
) -> PyResult<Bound<'py, pyo3::types::PyList>> {
    match series_type {
        SeriesType::Pandas => {
            // Try to_list() first, fallback to tolist()
            if series.hasattr("to_list")? {
                let list = series.call_method0("to_list")?;
                Ok(list.cast::<pyo3::types::PyList>()?.clone())
            } else {
                let list = series.call_method0("tolist")?;
                Ok(list.cast::<pyo3::types::PyList>()?.clone())
            }
        }

        SeriesType::Polars => {
            // Polars uses to_list()
            let list = series.call_method0("to_list")?;
            Ok(list.cast::<pyo3::types::PyList>()?.clone())
        }

        SeriesType::PyArrow => {
            // PyArrow Arrays use to_pylist()
            let list = series.call_method0("to_pylist")?;
            Ok(list.cast::<pyo3::types::PyList>()?.clone())
        }

        SeriesType::PySpark => {
            // PySpark doesn't have Series, but if it exists, convert via pandas
            let pandas_series = series.call_method0("toPandas")?;
            let list = pandas_series.call_method0("tolist")?;
            Ok(list.cast::<pyo3::types::PyList>()?.clone())
        }

        SeriesType::Generic => {
            // Try to_list() first, fallback to tolist()
            if series.hasattr("to_list")? {
                let list = series.call_method0("to_list")?;
                Ok(list.cast::<pyo3::types::PyList>()?.clone())
            } else if series.hasattr("tolist")? {
                let list = series.call_method0("tolist")?;
                Ok(list.cast::<pyo3::types::PyList>()?.clone())
            } else {
                Err(JsonToolsError::new_err(
                    "Generic Series must have to_list() or tolist() method",
                ))
            }
        }
    }
}

/// Helper macro to reduce boilerplate in PyJSONTools builder methods.
/// Each builder method follows the same pattern: lock the mutex, take the inner
/// JSONTools, apply a builder method, store the result back, and return `slf`.
#[cfg(feature = "python")]
macro_rules! py_builder_method {
    ($slf:expr, $tools:ident, $body:expr) => {{
        let mut guard = lock_config(&$slf.inner)?;
        let $tools = mem::take(&mut *guard);
        *guard = $body;
        drop(guard);
        Ok($slf)
    }};
}

#[cfg(feature = "python")]
#[pymethods]
impl PyJSONTools {
    /// Create a new JSONTools instance with default settings.
    ///
    /// Returns a new builder that can be configured with flatten(), unflatten(),
    /// or normal() mode, optional transformations, and then executed.
    ///
    /// Example:
    ///     >>> tools = JSONTools()
    ///     >>> result = tools.flatten().execute({"user": {"name": "John"}})
    #[new]
    #[pyo3(text_signature = "()")]
    pub fn new() -> Self {
        Self {
            inner: Mutex::new(JSONTools::new()),
        }
    }

    /// Set operation mode to flatten nested JSON into dot-separated keys.
    ///
    /// Example:
    ///     >>> result = JSONTools().flatten().execute({"a": {"b": 1}})
    ///     >>> result == {"a.b": 1}
    #[pyo3(text_signature = "($self)")]
    #[inline]
    pub fn flatten(slf: PyRef<'_, Self>) -> PyResult<PyRef<'_, Self>> {
        py_builder_method!(slf, tools, tools.flatten())
    }

    /// Set operation mode to unflatten dot-separated keys back into nested JSON.
    ///
    /// Example:
    ///     >>> result = JSONTools().unflatten().execute({"a.b": 1})
    ///     >>> result == {"a": {"b": 1}}
    #[pyo3(text_signature = "($self)")]
    #[inline]
    pub fn unflatten(slf: PyRef<'_, Self>) -> PyResult<PyRef<'_, Self>> {
        py_builder_method!(slf, tools, tools.unflatten())
    }

    /// Set operation mode to normal (apply transformations without flatten/unflatten).
    ///
    /// In normal mode, key/value replacements, filtering, and type conversion
    /// are applied recursively without changing the nesting structure.
    #[pyo3(text_signature = "($self)")]
    #[inline]
    pub fn normal(slf: PyRef<'_, Self>) -> PyResult<PyRef<'_, Self>> {
        py_builder_method!(slf, tools, tools.normal())
    }

    /// Set the separator for nested keys (default: ".").
    ///
    /// Args:
    ///     separator: Non-empty string to use between nested key segments.
    ///
    /// Raises:
    ///     ValueError: If separator is empty.
    #[pyo3(text_signature = "($self, separator)")]
    #[inline]
    pub fn separator(slf: PyRef<'_, Self>, separator: String) -> PyResult<PyRef<'_, Self>> {
        if separator.is_empty() {
            return Err(PyValueError::new_err("Separator cannot be empty"));
        }
        py_builder_method!(slf, tools, tools.separator(separator))
    }

    /// Enable or disable lowercase key conversion.
    ///
    /// Args:
    ///     value: True to convert all keys to lowercase.
    #[pyo3(text_signature = "($self, value)")]
    #[inline]
    pub fn lowercase_keys(slf: PyRef<'_, Self>, value: bool) -> PyResult<PyRef<'_, Self>> {
        py_builder_method!(slf, tools, tools.lowercase_keys(value))
    }

    /// Enable or disable removal of keys with empty string values.
    ///
    /// Args:
    ///     value: True to remove keys whose values are empty strings ("").
    #[pyo3(text_signature = "($self, value)")]
    #[inline]
    pub fn remove_empty_strings(slf: PyRef<'_, Self>, value: bool) -> PyResult<PyRef<'_, Self>> {
        py_builder_method!(slf, tools, tools.remove_empty_strings(value))
    }

    /// Enable or disable removal of keys with null values.
    ///
    /// Args:
    ///     value: True to remove keys whose values are null/None.
    #[pyo3(text_signature = "($self, value)")]
    #[inline]
    pub fn remove_nulls(slf: PyRef<'_, Self>, value: bool) -> PyResult<PyRef<'_, Self>> {
        py_builder_method!(slf, tools, tools.remove_nulls(value))
    }

    /// Enable or disable removal of keys with empty object values ({}).
    ///
    /// Args:
    ///     value: True to remove keys whose values are empty objects.
    #[pyo3(text_signature = "($self, value)")]
    #[inline]
    pub fn remove_empty_objects(slf: PyRef<'_, Self>, value: bool) -> PyResult<PyRef<'_, Self>> {
        py_builder_method!(slf, tools, tools.remove_empty_objects(value))
    }

    /// Enable or disable removal of keys with empty array values ([]).
    ///
    /// Args:
    ///     value: True to remove keys whose values are empty arrays.
    #[pyo3(text_signature = "($self, value)")]
    #[inline]
    pub fn remove_empty_arrays(slf: PyRef<'_, Self>, value: bool) -> PyResult<PyRef<'_, Self>> {
        py_builder_method!(slf, tools, tools.remove_empty_arrays(value))
    }

    /// Add a key replacement pattern (regex or literal fallback).
    ///
    /// Args:
    ///     find: Regex pattern to match against keys. Falls back to literal
    ///         string replacement if regex compilation fails.
    ///     replace: Replacement string.
    #[pyo3(text_signature = "($self, find, replace)")]
    #[inline]
    pub fn key_replacement(
        slf: PyRef<'_, Self>,
        find: String,
        replace: String,
    ) -> PyResult<PyRef<'_, Self>> {
        py_builder_method!(slf, tools, tools.key_replacement(find, replace))
    }

    /// Add a value replacement pattern (regex or literal fallback).
    ///
    /// Args:
    ///     find: Regex pattern to match against values. Falls back to literal
    ///         string replacement if regex compilation fails.
    ///     replace: Replacement string.
    #[pyo3(text_signature = "($self, find, replace)")]
    #[inline]
    pub fn value_replacement(
        slf: PyRef<'_, Self>,
        find: String,
        replace: String,
    ) -> PyResult<PyRef<'_, Self>> {
        py_builder_method!(slf, tools, tools.value_replacement(find, replace))
    }

    /// Enable collision handling by collecting duplicate keys into arrays.
    ///
    /// When key transformations produce duplicate keys, values are collected
    /// into arrays (e.g., "name": ["John", "Jane"]).
    ///
    /// Args:
    ///     value: True to enable collision handling.
    #[pyo3(text_signature = "($self, value)")]
    #[inline]
    pub fn handle_key_collision(slf: PyRef<'_, Self>, value: bool) -> PyResult<PyRef<'_, Self>> {
        py_builder_method!(slf, tools, tools.handle_key_collision(value))
    }

    /// Enable automatic type conversion from strings to numbers and booleans
    ///
    /// When enabled, the library will attempt to convert string values:
    /// - Numbers: "123" → 123, "1,234.56" → 1234.56, "$99.99" → 99.99
    /// - Booleans: "true"/"TRUE"/"True" → true, "false"/"FALSE"/"False" → false
    ///
    /// If conversion fails, the original string value is kept.
    ///
    /// # Arguments
    /// * `enable` - Whether to enable automatic type conversion
    ///
    /// # Returns
    /// Self for method chaining
    ///
    /// # Example
    /// ```python
    /// import json_tools_rs as jt
    /// result = jt.JSONTools().flatten().auto_convert_types(True).execute({"id": "123", "active": "true"})
    /// print(result)  # {'id': 123, 'active': True}
    /// ```
    ///
    #[pyo3(text_signature = "($self, enable)")]
    #[inline]
    pub fn auto_convert_types(slf: PyRef<'_, Self>, enable: bool) -> PyResult<PyRef<'_, Self>> {
        py_builder_method!(slf, tools, tools.auto_convert_types(enable))
    }

    /// Set the minimum batch size for parallel processing
    ///
    /// When processing multiple JSON documents, this threshold determines when to use
    /// parallel processing. Batches smaller than this threshold will be processed sequentially
    /// to avoid the overhead of thread spawning.
    ///
    /// Default: 10 items (can be overridden with JSON_TOOLS_PARALLEL_THRESHOLD environment variable)
    ///
    /// # Arguments
    /// * `threshold` - Minimum number of items in a batch to trigger parallel processing
    ///
    /// # Returns
    /// Self for method chaining
    ///
    /// # Example
    /// ```python
    /// import json_tools_rs as jt
    /// # Only use parallelism for batches of 50+ items
    /// tools = jt.JSONTools().flatten().parallel_threshold(50)
    /// results = tools.execute([...])  # Large batch will use parallel processing
    /// ```
    ///
    #[pyo3(text_signature = "($self, threshold)")]
    #[inline]
    pub fn parallel_threshold(slf: PyRef<'_, Self>, threshold: usize) -> PyResult<PyRef<'_, Self>> {
        py_builder_method!(slf, tools, tools.parallel_threshold(threshold))
    }

    /// Configure the number of threads for parallel processing
    ///
    /// By default, the number of logical CPUs is used. This method allows you to override
    /// that behavior for specific workloads or resource constraints.
    ///
    /// # Arguments
    /// * `num_threads` - Number of threads to use (None = use system default)
    ///
    /// # Returns
    /// Self for method chaining
    ///
    /// # Example
    /// ```python
    /// import json_tools_rs as jt
    /// # Limit to 4 threads for resource-constrained environments
    /// tools = jt.JSONTools().flatten().num_threads(4)
    /// # Or use None to auto-detect (default)
    /// tools = jt.JSONTools().flatten().num_threads(None)
    /// ```
    ///
    #[pyo3(text_signature = "($self, num_threads)")]
    #[inline]
    pub fn num_threads(
        slf: PyRef<'_, Self>,
        num_threads: Option<usize>,
    ) -> PyResult<PyRef<'_, Self>> {
        py_builder_method!(slf, tools, tools.num_threads(num_threads))
    }

    /// Configure the threshold for nested parallel processing within individual JSON documents
    ///
    /// When flattening or unflattening a single large JSON document, this threshold determines
    /// when to parallelize the processing of objects and arrays. Only objects/arrays with more
    /// than this many keys/items will be processed in parallel.
    ///
    /// Default: 100 (can be overridden with JSON_TOOLS_NESTED_PARALLEL_THRESHOLD environment variable)
    ///
    /// # Arguments
    /// * `threshold` - Minimum number of keys/items to trigger nested parallelism
    ///
    /// # Returns
    /// Self for method chaining
    ///
    /// # Example
    /// ```python
    /// import json_tools_rs as jt
    /// # Only parallelize objects/arrays with 200+ items
    /// tools = jt.JSONTools().flatten().nested_parallel_threshold(200)
    /// result = tools.execute(large_json)  # Large nested structures will use parallel processing
    /// ```
    ///
    #[pyo3(text_signature = "($self, threshold)")]
    #[inline]
    pub fn nested_parallel_threshold(
        slf: PyRef<'_, Self>,
        threshold: usize,
    ) -> PyResult<PyRef<'_, Self>> {
        py_builder_method!(slf, tools, tools.nested_parallel_threshold(threshold))
    }

    /// Set the maximum array index allowed during unflattening (DoS protection)
    ///
    /// Prevents malicious flattened keys like "items.999999999" from causing
    /// excessive memory allocation. Keys with array indices exceeding this
    /// limit will produce an error during unflattening.
    ///
    /// Default: 100,000
    #[pyo3(text_signature = "($self, max)")]
    #[inline]
    pub fn max_array_index(slf: PyRef<'_, Self>, max: usize) -> PyResult<PyRef<'_, Self>> {
        py_builder_method!(slf, tools, tools.max_array_index(max))
    }

    /// Execute the configured JSON operation
    ///
    /// This method executes the configured operation (flatten or unflatten) with all
    /// the specified transformations, collision handling, and filtering options.
    ///
    /// # Arguments
    /// * `json_input` - JSON input as:
    ///   - str: JSON string
    ///   - dict: Python dictionary (will be serialized to JSON)
    ///   - list[str]: List of JSON strings
    ///   - list[dict]: List of Python dictionaries (will be serialized to JSON)
    ///
    /// # Returns
    /// * str input → str output (processed JSON string)
    /// * dict input → dict output (processed Python dictionary)
    /// * list[str] input → list[str] output (list of processed JSON strings)
    /// * list[dict] input → list[dict] output (list of processed Python dictionaries)
    ///
    /// # Performance
    /// Uses interior mutability to avoid cloning JSONTools - only clones for execute() call
    #[pyo3(text_signature = "($self, json_input)")]
    pub fn execute(&self, json_input: &Bound<'_, PyAny>) -> PyResult<Py<PyAny>> {
        let py = json_input.py();

        // NEW: Check for DataFrame or Series first (before other type checks)
        match detect_data_structure(json_input)? {
            Some(DataStructureType::DataFrame(df_type)) => {
                return self.execute_dataframe(json_input, df_type);
            }
            Some(DataStructureType::Series(series_type)) => {
                return self.execute_series(json_input, series_type);
            }
            None => {
                // Fall through to existing type checks
            }
        }

        // Fast path: single JSON string → return JSON string
        if let Ok(json_str) = json_input.extract::<String>() {
            // TIER 6→3 OPTIMIZATION: Take ownership instead of cloning
            // Saves 1K-10K cycles by avoiding deep clone of entire JSONTools config
            let result = py
                .detach(|| {
                    let mut guard = lock_config(&self.inner)?;
                    let tools = mem::take(&mut *guard);
                    let result = tools.execute(json_str.as_str());
                    *guard = tools;
                    result
                })
                .map_err(|e| {
                    JsonToolsError::new_err(format!("Failed to process JSON string: {}", e))
                })?;

            match result {
                JsonOutput::Single(processed) => {
                    Ok(processed.into_pyobject(py)?.into_any().unbind())
                }
                JsonOutput::Multiple(_) => Err(PyValueError::new_err(
                    "Unexpected multiple results for single JSON input",
                )),
            }
        } else if json_input.is_instance_of::<pyo3::types::PyDict>() {
            // TIER 6→3 OPTIMIZATION: Direct Python dict → serde_json::Value conversion
            // Eliminates expensive Python json.dumps() + json.loads() round-trip
            // Saves 50K-500K cycles per dict!

            // Convert Python dict → serde_json::Value (direct, no JSON string intermediate)
            let value: serde_json::Value = depythonize(json_input).map_err(|e| {
                JsonToolsError::new_err(format!("Failed to convert Python dict: {}", e))
            })?;

            // Serialize to JSON string using fast sonic-rs
            let json_str = json_parser::to_string(&value)
                .map_err(|e| JsonToolsError::new_err(format!("Failed to serialize: {}", e)))?;

            // Process with Rust tools (release GIL)
            let result = py
                .detach(|| {
                    let mut guard = lock_config(&self.inner)?;
                    let tools = mem::take(&mut *guard);
                    let result = tools.execute(json_str.as_str());
                    *guard = tools;
                    result
                })
                .map_err(|e| {
                    JsonToolsError::new_err(format!("Failed to process Python dict: {}", e))
                })?;

            match result {
                JsonOutput::Single(processed) => {
                    // Parse result and convert back to Python dict (direct, no json.loads!)
                    let result_value: serde_json::Value = json_parser::from_str(&processed)
                        .map_err(|e| {
                            JsonToolsError::new_err(format!("Failed to parse result: {}", e))
                        })?;

                    let python_dict = pythonize(py, &result_value).map_err(|e| {
                        JsonToolsError::new_err(format!("Failed to convert to Python: {}", e))
                    })?;

                    Ok(python_dict.unbind())
                }
                JsonOutput::Multiple(_) => Err(PyValueError::new_err(
                    "Unexpected multiple results for single dict input",
                )),
            }
        } else if json_input.is_instance_of::<pyo3::types::PyList>() {
            // Handle list input - batch processing of JSON strings and/or dicts
            let list = json_input.cast::<pyo3::types::PyList>()?;

            if list.is_empty() {
                return Ok(Vec::<String>::new().into_pyobject(py)?.into_any().unbind());
            }

            // TIER 6→3 OPTIMIZATION: Use pythonize for batch dict conversion
            // Avoids expensive Python json.dumps() for each dict
            let mut json_strings: Vec<String> = Vec::with_capacity(list.len());
            let mut is_str_flags: Vec<bool> = Vec::with_capacity(list.len());
            let mut has_other_types = false;

            for item in list.iter() {
                if let Ok(json_str) = item.extract::<String>() {
                    json_strings.push(json_str);
                    is_str_flags.push(true);
                } else if item.is_instance_of::<pyo3::types::PyDict>() {
                    // Direct conversion: Python dict → serde_json::Value → JSON string
                    let value: serde_json::Value = depythonize(&item).map_err(|e| {
                        JsonToolsError::new_err(format!("Failed to convert dict in list: {}", e))
                    })?;
                    let json_str = json_parser::to_string(&value).map_err(|e| {
                        JsonToolsError::new_err(format!("Failed to serialize dict: {}", e))
                    })?;
                    json_strings.push(json_str);
                    is_str_flags.push(false);
                } else {
                    has_other_types = true;
                    break;
                }
            }

            if has_other_types {
                return Err(PyValueError::new_err(
                    "List items must be either JSON strings or Python dictionaries",
                ));
            }

            // TIER 6→3 OPTIMIZATION: Take ownership instead of cloning
            let result = py
                .detach(|| {
                    let mut guard = lock_config(&self.inner)?;
                    let tools = mem::take(&mut *guard);
                    let result = tools.execute(json_strings);
                    *guard = tools;
                    result
                })
                .map_err(|e| {
                    JsonToolsError::new_err(format!("Failed to process JSON list: {}", e))
                })?;

            match result {
                JsonOutput::Single(_) => Err(PyValueError::new_err(
                    "Unexpected single result for multiple input",
                )),
                JsonOutput::Multiple(processed_list) => {
                    // Determine output shape and transform accordingly
                    let all_strings = is_str_flags.iter().all(|&b| b);
                    let all_dicts = is_str_flags.iter().all(|&b| !b);

                    if all_strings {
                        Ok(processed_list.into_pyobject(py)?.into_any().unbind())
                    } else if all_dicts {
                        // TIER 6→3: Direct conversion using pythonize (no Python json.loads!)
                        let mut dict_results: Vec<Py<PyAny>> =
                            Vec::with_capacity(processed_list.len());
                        for processed_json in processed_list {
                            let result_value: serde_json::Value =
                                json_parser::from_str(&processed_json).map_err(|e| {
                                    JsonToolsError::new_err(format!("Failed to parse JSON: {}", e))
                                })?;
                            let python_dict = pythonize(py, &result_value).map_err(|e| {
                                JsonToolsError::new_err(format!(
                                    "Failed to convert to Python dict: {}",
                                    e
                                ))
                            })?;
                            dict_results.push(python_dict.unbind());
                        }
                        Ok(dict_results.into_pyobject(py)?.into_any().unbind())
                    } else {
                        // TIER 6→3: Mixed results - use pythonize for dicts
                        let mut mixed_results: Vec<Py<PyAny>> =
                            Vec::with_capacity(processed_list.len());
                        for (processed_json, is_str) in processed_list.into_iter().zip(is_str_flags)
                        {
                            if is_str {
                                mixed_results
                                    .push(processed_json.into_pyobject(py)?.into_any().unbind());
                            } else {
                                let result_value: serde_json::Value =
                                    json_parser::from_str(&processed_json).map_err(|e| {
                                        JsonToolsError::new_err(format!(
                                            "Failed to parse JSON: {}",
                                            e
                                        ))
                                    })?;
                                let python_dict = pythonize(py, &result_value).map_err(|e| {
                                    JsonToolsError::new_err(format!(
                                        "Failed to convert to Python dict: {}",
                                        e
                                    ))
                                })?;
                                mixed_results.push(python_dict.unbind());
                            }
                        }
                        Ok(mixed_results.into_pyobject(py)?.into_any().unbind())
                    }
                }
            }
        } else {
            Err(PyValueError::new_err(
                "json_input must be a JSON string, Python dict, list of JSON strings, or list of Python dicts",
            ))
        }
    }

    /// Execute the configured operation and return a JsonOutput object
    ///
    /// This method returns the full JsonOutput object for advanced use cases
    /// where you need to check the result type or handle both single and multiple
    /// results in a unified way.
    ///
    /// # Arguments
    /// * `json_input` - JSON input as:
    ///   - str: JSON string
    ///   - dict: Python dictionary (will be serialized to JSON)
    ///   - list[str]: List of JSON strings
    ///   - list[dict]: List of Python dictionaries (will be serialized to JSON)
    ///
    /// # Returns
    /// * `PyJsonOutput` - JsonOutput object with is_single/is_multiple methods
    ///
    /// # Performance
    /// Uses interior mutability to avoid cloning JSONTools - only clones for execute() call
    #[pyo3(text_signature = "($self, json_input)")]
    pub fn execute_to_output(&self, json_input: &Bound<'_, PyAny>) -> PyResult<PyJsonOutput> {
        let py = json_input.py();

        // Note: DataFrames/Series are not supported in execute_to_output()
        // Use execute() instead for DataFrame/Series support

        // Single JSON string
        if let Ok(json_str) = json_input.extract::<String>() {
            // TIER 6→3: Take ownership instead of cloning (10K-50K cycles saved)
            let result = py
                .detach(|| {
                    let mut guard = lock_config(&self.inner)?;
                    let tools = mem::take(&mut *guard);
                    let result = tools.execute(json_str.as_str());
                    *guard = tools;
                    result
                })
                .map_err(|e| {
                    JsonToolsError::new_err(format!("Failed to process JSON string: {}", e))
                })?;
            return Ok(PyJsonOutput::from_rust_output(result));
        }

        // Single Python dictionary - use pythonize for direct conversion
        if json_input.is_instance_of::<pyo3::types::PyDict>() {
            // TIER 6→3: Direct Python dict → serde_json::Value conversion
            let value: serde_json::Value = depythonize(json_input).map_err(|e| {
                JsonToolsError::new_err(format!("Failed to convert Python dict: {}", e))
            })?;

            let json_str = json_parser::to_string(&value)
                .map_err(|e| JsonToolsError::new_err(format!("Failed to serialize: {}", e)))?;

            // TIER 6→3 OPTIMIZATION: Take ownership instead of cloning
            let result = py
                .detach(|| {
                    let mut guard = lock_config(&self.inner)?;
                    let tools = mem::take(&mut *guard);
                    let result = tools.execute(json_str.as_str());
                    *guard = tools;
                    result
                })
                .map_err(|e| {
                    JsonToolsError::new_err(format!("Failed to process Python dict: {}", e))
                })?;
            return Ok(PyJsonOutput::from_rust_output(result));
        }

        // List input - batch processing or single JSON array
        if json_input.is_instance_of::<pyo3::types::PyList>() {
            let list = json_input.cast::<pyo3::types::PyList>()?;

            if list.is_empty() {
                return Ok(PyJsonOutput::from_rust_output(JsonOutput::Multiple(vec![])));
            }

            // TIER 6→3: Serialize inputs using pythonize for dicts
            let mut json_strings: Vec<String> = Vec::with_capacity(list.len());

            for item in list.iter() {
                if let Ok(json_str) = item.extract::<String>() {
                    json_strings.push(json_str);
                } else if item.is_instance_of::<pyo3::types::PyDict>() {
                    // Direct conversion: Python dict → serde_json::Value → JSON string
                    let value: serde_json::Value = depythonize(&item).map_err(|e| {
                        JsonToolsError::new_err(format!("Failed to convert dict in list: {}", e))
                    })?;
                    let json_str = json_parser::to_string(&value).map_err(|e| {
                        JsonToolsError::new_err(format!("Failed to serialize dict: {}", e))
                    })?;
                    json_strings.push(json_str);
                } else {
                    return Err(PyValueError::new_err(
                        "List items must be either JSON strings or Python dictionaries",
                    ));
                }
            }

            // Process the list of JSON strings directly
            // TIER 6→3: Take ownership instead of cloning (10K-50K cycles saved)
            let result = py
                .detach(|| {
                    let mut guard = lock_config(&self.inner)?;
                    let tools = mem::take(&mut *guard);
                    let result = tools.execute(json_strings);
                    *guard = tools;
                    result
                })
                .map_err(|e| {
                    JsonToolsError::new_err(format!("Failed to process JSON list: {}", e))
                })?;
            return Ok(PyJsonOutput::from_rust_output(result));
        }

        Err(PyValueError::new_err(
            "json_input must be a JSON string, Python dict, DataFrame, Series, list of JSON strings, or list of Python dicts",
        ))
    }
}

// =============================================================================
// DataFrame Reconstruction Functions
// =============================================================================

/// Reconstruct DataFrame from list of dicts (with fallback to list)
#[cfg(feature = "python")]
fn reconstruct_dataframe(
    py: Python,
    df_type: DataFrameType,
    processed_dicts: Vec<Py<PyAny>>,
) -> PyResult<Py<PyAny>> {
    match df_type {
        DataFrameType::Pandas => reconstruct_pandas_df(py, processed_dicts),
        DataFrameType::Polars => reconstruct_polars_df(py, processed_dicts),
        DataFrameType::PyArrow => reconstruct_pyarrow_table(py, processed_dicts),
        DataFrameType::PySpark => {
            // PySpark reconstruction would need SparkSession - fallback to list for now
            Ok(processed_dicts.into_pyobject(py)?.into_any().unbind())
        }
        DataFrameType::Generic => {
            // Generic - just return list of dicts
            Ok(processed_dicts.into_pyobject(py)?.into_any().unbind())
        }
    }
}

/// Reconstruct pandas DataFrame
#[cfg(feature = "python")]
fn reconstruct_pandas_df(py: Python, records: Vec<Py<PyAny>>) -> PyResult<Py<PyAny>> {
    // Convert to Python list once — O(1) refcount clone instead of O(N) item clones
    let py_list = records.into_pyobject(py)?.into_any().unbind();

    match py.import("pandas") {
        Ok(pandas) => match pandas.call_method1("DataFrame", (py_list.clone_ref(py),)) {
            Ok(df) => Ok(df.unbind()),
            Err(_) => Ok(py_list),
        },
        Err(_) => Ok(py_list),
    }
}

/// Reconstruct polars DataFrame
#[cfg(feature = "python")]
fn reconstruct_polars_df(py: Python, records: Vec<Py<PyAny>>) -> PyResult<Py<PyAny>> {
    // Convert to Python list once — O(1) refcount clone instead of O(N) item clones
    let py_list = records.into_pyobject(py)?.into_any().unbind();

    match py.import("polars") {
        Ok(polars) => match polars.call_method1("DataFrame", (py_list.clone_ref(py),)) {
            Ok(df) => Ok(df.unbind()),
            Err(_) => Ok(py_list),
        },
        Err(_) => Ok(py_list),
    }
}

/// Reconstruct PyArrow Table
#[cfg(feature = "python")]
fn reconstruct_pyarrow_table(py: Python, records: Vec<Py<PyAny>>) -> PyResult<Py<PyAny>> {
    // Convert to Python list once — O(1) refcount clone instead of O(N) item clones
    let py_list = records.into_pyobject(py)?.into_any().unbind();

    match py.import("pyarrow") {
        Ok(pyarrow) => {
            let table_class = pyarrow.getattr("Table")?;
            match table_class.call_method1("from_pylist", (py_list.clone_ref(py),)) {
                Ok(table) => Ok(table.unbind()),
                Err(_) => Ok(py_list),
            }
        }
        Err(_) => Ok(py_list),
    }
}

// =============================================================================
// Series Reconstruction Functions
// =============================================================================

/// Reconstruct Series from list (with fallback to list)
#[cfg(feature = "python")]
fn reconstruct_series(
    py: Python,
    series_type: SeriesType,
    processed_items: Vec<Py<PyAny>>,
) -> PyResult<Py<PyAny>> {
    match series_type {
        SeriesType::Pandas => reconstruct_pandas_series(py, processed_items),
        SeriesType::Polars => reconstruct_polars_series(py, processed_items),
        SeriesType::PyArrow => reconstruct_pyarrow_array(py, processed_items),
        SeriesType::PySpark => {
            // PySpark doesn't have Series - fallback to list
            Ok(processed_items.into_pyobject(py)?.into_any().unbind())
        }
        SeriesType::Generic => {
            // Generic - just return list
            Ok(processed_items.into_pyobject(py)?.into_any().unbind())
        }
    }
}

/// Reconstruct pandas Series
#[cfg(feature = "python")]
fn reconstruct_pandas_series(py: Python, items: Vec<Py<PyAny>>) -> PyResult<Py<PyAny>> {
    // Convert to Python list once — O(1) refcount clone instead of O(N) item clones
    let py_list = items.into_pyobject(py)?.into_any().unbind();

    match py.import("pandas") {
        Ok(pandas) => match pandas.call_method1("Series", (py_list.clone_ref(py),)) {
            Ok(series) => Ok(series.unbind()),
            Err(_) => Ok(py_list),
        },
        Err(_) => Ok(py_list),
    }
}

/// Reconstruct polars Series
#[cfg(feature = "python")]
fn reconstruct_polars_series(py: Python, items: Vec<Py<PyAny>>) -> PyResult<Py<PyAny>> {
    // Convert to Python list once — O(1) refcount clone instead of O(N) item clones
    let py_list = items.into_pyobject(py)?.into_any().unbind();

    match py.import("polars") {
        Ok(polars) => match polars.call_method1("Series", (py_list.clone_ref(py),)) {
            Ok(series) => Ok(series.unbind()),
            Err(_) => Ok(py_list),
        },
        Err(_) => Ok(py_list),
    }
}

/// Reconstruct PyArrow Array
#[cfg(feature = "python")]
fn reconstruct_pyarrow_array(py: Python, items: Vec<Py<PyAny>>) -> PyResult<Py<PyAny>> {
    // Convert to Python list once — O(1) refcount clone instead of O(N) item clones
    let py_list = items.into_pyobject(py)?.into_any().unbind();

    match py.import("pyarrow") {
        Ok(pyarrow) => match pyarrow.call_method1("array", (py_list.clone_ref(py),)) {
            Ok(array) => Ok(array.unbind()),
            Err(_) => Ok(py_list),
        },
        Err(_) => Ok(py_list),
    }
}

// =============================================================================
// PyJSONTools Helper Methods (DataFrame and Series Processing)
// =============================================================================

#[cfg(feature = "python")]
impl PyJSONTools {
    /// Process DataFrame through existing pipeline
    fn execute_dataframe(
        &self,
        df: &Bound<'_, PyAny>,
        df_type: DataFrameType,
    ) -> PyResult<Py<PyAny>> {
        let py = df.py();

        // Step 1: Convert DataFrame to list of dicts (as serde_json::Value)
        let records = dataframe_to_records(df, df_type)?;

        // Step 2: Serialize to JSON strings for processing
        let mut json_strings = Vec::with_capacity(records.len());
        for record in records {
            let json_str = json_parser::to_string(&record).map_err(|e| {
                JsonToolsError::new_err(format!("Failed to serialize record: {}", e))
            })?;
            json_strings.push(json_str);
        }

        // Step 3: Process through existing pipeline (releases GIL)
        let result = py
            .detach(|| {
                let mut guard = lock_config(&self.inner)?;
                let tools = mem::take(&mut *guard);
                let result = tools.execute(json_strings); // Batch processing
                *guard = tools;
                result
            })
            .map_err(|e| JsonToolsError::new_err(format!("Failed to process DataFrame: {}", e)))?;

        // Step 4: Reconstruct DataFrame from results
        match result {
            JsonOutput::Multiple(processed_list) => {
                // Convert JSON strings back to Python dicts
                let mut processed_dicts: Vec<Py<PyAny>> = Vec::with_capacity(processed_list.len());
                for json_str in processed_list {
                    let value: serde_json::Value =
                        json_parser::from_str(&json_str).map_err(|e| {
                            JsonToolsError::new_err(format!("Failed to parse result: {}", e))
                        })?;
                    let py_dict = pythonize(py, &value).map_err(|e| {
                        JsonToolsError::new_err(format!("Failed to convert to Python: {}", e))
                    })?;
                    processed_dicts.push(py_dict.unbind());
                }

                // Reconstruct DataFrame (with fallback to list)
                reconstruct_dataframe(py, df_type, processed_dicts)
            }
            JsonOutput::Single(_) => Err(PyValueError::new_err(
                "Unexpected single result for DataFrame input",
            )),
        }
    }

    /// Process Series through existing list pipeline (REUSE existing code!)
    fn execute_series(
        &self,
        series: &Bound<'_, PyAny>,
        series_type: SeriesType,
    ) -> PyResult<Py<PyAny>> {
        let py = series.py();

        // Step 1: Convert Series to Python list
        let list = series_to_list(series, series_type)?;

        // Step 2: Process using EXISTING list handling code (copy from execute() method)
        let mut json_strings: Vec<String> = Vec::with_capacity(list.len());
        let mut is_str_flags: Vec<bool> = Vec::with_capacity(list.len());
        let mut has_other_types = false;

        for item in list.iter() {
            if let Ok(json_str) = item.extract::<String>() {
                json_strings.push(json_str);
                is_str_flags.push(true);
            } else if item.is_instance_of::<pyo3::types::PyDict>() {
                // Direct conversion: Python dict → serde_json::Value → JSON string
                let value: serde_json::Value = depythonize(&item).map_err(|e| {
                    JsonToolsError::new_err(format!("Failed to convert dict in series: {}", e))
                })?;
                let json_str = json_parser::to_string(&value).map_err(|e| {
                    JsonToolsError::new_err(format!("Failed to serialize dict: {}", e))
                })?;
                json_strings.push(json_str);
                is_str_flags.push(false);
            } else {
                has_other_types = true;
                break;
            }
        }

        if has_other_types {
            return Err(PyValueError::new_err(
                "Series items must be either JSON strings or Python dictionaries",
            ));
        }

        // Step 3: Process through existing pipeline (releases GIL)
        let result = py
            .detach(|| {
                let mut guard = lock_config(&self.inner)?;
                let tools = mem::take(&mut *guard);
                let result = tools.execute(json_strings);
                *guard = tools;
                result
            })
            .map_err(|e| JsonToolsError::new_err(format!("Failed to process Series: {}", e)))?;

        // Step 4: Reconstruct Series from results
        match result {
            JsonOutput::Multiple(processed_list) => {
                // Type preservation: convert back to appropriate format
                let all_strings = is_str_flags.iter().all(|&b| b);
                let all_dicts = is_str_flags.iter().all(|&b| !b);

                let processed_items: Vec<Py<PyAny>> = if all_strings {
                    // All strings - convert to list of strings
                    processed_list
                        .into_iter()
                        .map(|s| {
                            s.into_pyobject(py)
                                .map(|o| o.into_any().unbind())
                                .map_err(|e| {
                                    JsonToolsError::new_err(format!(
                                        "Failed to convert string to Python object: {}",
                                        e
                                    ))
                                })
                        })
                        .collect::<PyResult<Vec<_>>>()?
                } else if all_dicts {
                    // All dicts - convert to list of dicts
                    let mut dict_results = Vec::with_capacity(processed_list.len());
                    for processed_json in processed_list {
                        let result_value: serde_json::Value =
                            json_parser::from_str(&processed_json).map_err(|e| {
                                JsonToolsError::new_err(format!("Failed to parse result: {}", e))
                            })?;
                        let python_dict = pythonize(py, &result_value).map_err(|e| {
                            JsonToolsError::new_err(format!(
                                "Failed to convert to Python dict: {}",
                                e
                            ))
                        })?;
                        dict_results.push(python_dict.unbind());
                    }
                    dict_results
                } else {
                    // Mixed results - convert each based on type
                    let mut mixed_results = Vec::with_capacity(processed_list.len());
                    for (processed_json, is_str) in processed_list.into_iter().zip(is_str_flags) {
                        if is_str {
                            mixed_results
                                .push(processed_json.into_pyobject(py)?.into_any().unbind());
                        } else {
                            let result_value: serde_json::Value =
                                json_parser::from_str(&processed_json).map_err(|e| {
                                    JsonToolsError::new_err(format!(
                                        "Failed to parse result: {}",
                                        e
                                    ))
                                })?;
                            let python_dict = pythonize(py, &result_value).map_err(|e| {
                                JsonToolsError::new_err(format!(
                                    "Failed to convert to Python: {}",
                                    e
                                ))
                            })?;
                            mixed_results.push(python_dict.unbind());
                        }
                    }
                    mixed_results
                };

                // Reconstruct Series (with fallback to list)
                reconstruct_series(py, series_type, processed_items)
            }
            JsonOutput::Single(_) => Err(PyValueError::new_err(
                "Unexpected single result for Series input",
            )),
        }
    }
}

/// Python module definition
#[cfg(feature = "python")]
#[pymodule]
fn json_tools_rs(m: &Bound<'_, PyModule>) -> PyResult<()> {
    // Add the unified JSONTools class
    m.add_class::<PyJSONTools>()?;

    // Add the JsonOutput class for results
    m.add_class::<PyJsonOutput>()?;

    // Add the custom exception
    m.add("JsonToolsError", m.py().get_type::<JsonToolsError>())?;

    // Add module metadata
    m.add("__version__", env!("CARGO_PKG_VERSION"))?;
    m.add("__author__", "JSON Tools RS Contributors")?;
    m.add(
        "__description__",
        "Python bindings for JSON Tools RS - Unified JSON manipulation library with advanced collision handling and filtering",
    )?;

    Ok(())
}