use std::num::NonZeroUsize;
use std::sync::Arc;
use std::sync::{Mutex, PoisonError};

use lru::LruCache;
use pyo3::IntoPyObjectExt;
use pyo3::intern;
use pyo3::prelude::*;
use pyo3::sync::MutexExt;
use pyo3::types::{PyDict, PyString};
use regex::Regex;

use crate::build_tools::LazyLock;
use crate::build_tools::{is_strict, py_schema_error_type, schema_or_config, schema_or_config_same};
use crate::errors::{ErrorType, ValError, ValResult};
use crate::input::Input;
use crate::tools::SchemaDict;

use super::{BuildValidator, CombinedValidator, DefinitionsBuilder, ValidationState, Validator};

#[derive(Debug)]
pub struct StrValidator {
    strict: bool,
    coerce_numbers_to_str: bool,
}

static STRICT_STR_VALIDATOR: LazyLock<Arc<CombinedValidator>> = LazyLock::new(|| {
    CombinedValidator::Str(StrValidator {
        strict: true,
        coerce_numbers_to_str: false,
    })
    .into()
});

static LAX_STR_VALIDATOR: LazyLock<Arc<CombinedValidator>> = LazyLock::new(|| {
    CombinedValidator::Str(StrValidator {
        strict: false,
        coerce_numbers_to_str: false,
    })
    .into()
});

impl BuildValidator for StrValidator {
    const EXPECTED_TYPE: &'static str = "str";

    fn build(
        schema: &Bound<'_, PyDict>,
        config: Option<&Bound<'_, PyDict>>,
        _definitions: &mut DefinitionsBuilder<Arc<CombinedValidator>>,
    ) -> PyResult<Arc<CombinedValidator>> {
        let con_str_validator = StrConstrainedValidator::build(schema, config)?;

        if con_str_validator.has_constraints_set() {
            Ok(Arc::new(con_str_validator.into()))
        } else if !con_str_validator.coerce_numbers_to_str {
            if is_strict(schema, config)? {
                Ok(STRICT_STR_VALIDATOR.clone())
            } else {
                Ok(LAX_STR_VALIDATOR.clone())
            }
        } else {
            Ok(CombinedValidator::Str(StrValidator {
                strict: con_str_validator.strict,
                coerce_numbers_to_str: con_str_validator.coerce_numbers_to_str,
            })
            .into())
        }
    }
}

impl_py_gc_traverse!(StrValidator {});

impl Validator for StrValidator {
    fn validate<'py>(
        &self,
        py: Python<'py>,
        input: &(impl Input<'py> + ?Sized),
        state: &mut ValidationState<'_, 'py>,
    ) -> ValResult<Py<PyAny>> {
        input
            .validate_str(state.strict_or(self.strict), self.coerce_numbers_to_str)
            .and_then(|val_match| {
                Ok(val_match
                    .unpack(state)
                    .as_py_string(py, state.cache_str())
                    .into_py_any(py)?)
            })
    }

    fn get_name(&self) -> &str {
        Self::EXPECTED_TYPE
    }
}

/// Any new properties set here must be reflected in `has_constraints_set`
#[derive(Debug, Clone, Default)]
pub struct StrConstrainedValidator {
    strict: bool,
    pattern: Option<Pattern>,
    max_length: Option<usize>,
    min_length: Option<usize>,
    strip_whitespace: bool,
    to_lower: bool,
    to_upper: bool,
    coerce_numbers_to_str: bool,
}

impl_py_gc_traverse!(StrConstrainedValidator {});

impl Validator for StrConstrainedValidator {
    fn validate<'py>(
        &self,
        py: Python<'py>,
        input: &(impl Input<'py> + ?Sized),
        state: &mut ValidationState<'_, 'py>,
    ) -> ValResult<Py<PyAny>> {
        let either_str = input
            .validate_str(state.strict_or(self.strict), self.coerce_numbers_to_str)?
            .unpack(state);
        let cow = either_str.as_cow()?;
        let mut str = cow.as_ref();
        if self.strip_whitespace {
            str = str.trim();
        }

        let str_len: Option<usize> = if self.min_length.is_some() | self.max_length.is_some() {
            Some(str.chars().count())
        } else {
            None
        };
        if let Some(min_length) = self.min_length
            && str_len.unwrap() < min_length
        {
            return Err(ValError::new(
                ErrorType::StringTooShort {
                    min_length,
                    context: None,
                },
                input,
            ));
        }
        if let Some(max_length) = self.max_length
            && str_len.unwrap() > max_length
        {
            return Err(ValError::new(
                ErrorType::StringTooLong {
                    max_length,
                    context: None,
                },
                input,
            ));
        }

        if let Some(pattern) = &self.pattern
            && !pattern.is_match(py, str)?
        {
            return Err(ValError::new(
                ErrorType::StringPatternMismatch {
                    pattern: pattern.pattern.clone(),
                    context: None,
                },
                input,
            ));
        }

        let py_string = if self.to_lower {
            state.maybe_cached_str(py, &str.to_lowercase())
        } else if self.to_upper {
            state.maybe_cached_str(py, &str.to_uppercase())
        } else if self.strip_whitespace {
            state.maybe_cached_str(py, str)
        } else {
            // we haven't modified the string, return the original as it might be a PyString
            either_str.as_py_string(py, state.cache_str())
        };
        Ok(py_string.into_py_any(py)?)
    }

    fn get_name(&self) -> &'static str {
        "constrained-str"
    }
}

impl StrConstrainedValidator {
    fn build(schema: &Bound<'_, PyDict>, config: Option<&Bound<'_, PyDict>>) -> PyResult<Self> {
        let py = schema.py();

        let pattern = schema
            .get_as(intern!(py, "pattern"))?
            .map(|s| {
                let regex_engine = schema_or_config::<Bound<'_, PyString>>(
                    schema,
                    config,
                    intern!(py, "regex_engine"),
                    intern!(py, "regex_engine"),
                )?;
                let regex_engine = regex_engine
                    .as_ref()
                    .map(|s| s.to_str())
                    .transpose()?
                    .unwrap_or(RegexEngine::RUST_REGEX);
                Pattern::compile(s, regex_engine)
            })
            .transpose()?;
        let min_length: Option<usize> =
            schema_or_config(schema, config, intern!(py, "min_length"), intern!(py, "str_min_length"))?;
        let max_length: Option<usize> =
            schema_or_config(schema, config, intern!(py, "max_length"), intern!(py, "str_max_length"))?;

        let strip_whitespace: bool = schema_or_config(
            schema,
            config,
            intern!(py, "strip_whitespace"),
            intern!(py, "str_strip_whitespace"),
        )?
        .unwrap_or(false);
        let to_lower: bool =
            schema_or_config(schema, config, intern!(py, "to_lower"), intern!(py, "str_to_lower"))?.unwrap_or(false);
        let to_upper: bool =
            schema_or_config(schema, config, intern!(py, "to_upper"), intern!(py, "str_to_upper"))?.unwrap_or(false);

        let coerce_numbers_to_str: bool =
            schema_or_config_same(schema, config, intern!(py, "coerce_numbers_to_str"))?.unwrap_or(false);

        Ok(Self {
            strict: is_strict(schema, config)?,
            pattern,
            min_length,
            max_length,
            strip_whitespace,
            to_lower,
            to_upper,
            coerce_numbers_to_str,
        })
    }

    // whether any of the constraints/customisations are actually enabled
    // except strict and coerce_numbers_to_str which can be set on StrValidator
    fn has_constraints_set(&self) -> bool {
        self.pattern.is_some()
            || self.max_length.is_some()
            || self.min_length.is_some()
            || self.strip_whitespace
            || self.to_lower
            || self.to_upper
    }
}

// A shared LRU cache for compiled regex objects.
// NOTE: We could push things even further by using an Arc over the `Regex` instances, but this is largely sufficient to avoid
// memory consumption issues.
static REGEX_CACHE: LazyLock<Mutex<LruCache<String, Regex>>> =
    LazyLock::new(|| Mutex::new(LruCache::new(NonZeroUsize::new(50).unwrap())));

#[derive(Debug, Clone)]
struct Pattern {
    pattern: String,
    engine: RegexEngine,
}

#[derive(Debug, Clone)]
enum RegexEngine {
    RustRegex(Regex),
    PythonRe(Py<PyAny>),
}

impl RegexEngine {
    const RUST_REGEX: &'static str = "rust-regex";
    const PYTHON_RE: &'static str = "python-re";
}

impl Pattern {
    fn extract_pattern_str(pattern: &Bound<'_, PyAny>) -> PyResult<String> {
        if pattern.is_instance_of::<PyString>() {
            Ok(pattern.to_string())
        } else {
            pattern
                .getattr("pattern")
                .and_then(|attr| attr.extract::<String>())
                .map_err(|_| py_schema_error_type!("Invalid pattern, must be str or re.Pattern: {pattern}"))
        }
    }

    fn compile(pattern: Bound<'_, PyAny>, engine: &str) -> PyResult<Self> {
        let pattern_str = Self::extract_pattern_str(&pattern)?;

        let py = pattern.py();

        let re_module = py.import(intern!(py, "re"))?;
        let re_compile = re_module.getattr(intern!(py, "compile"))?;
        let re_pattern = re_module.getattr(intern!(py, "Pattern"))?;

        if pattern.is_instance(&re_pattern)? {
            // if the pattern is already a compiled regex object, we default to using the python re engine
            // so that any flags, etc. are preserved
            Ok(Self {
                pattern: pattern_str,
                engine: RegexEngine::PythonRe(pattern.unbind()),
            })
        } else {
            let engine = match engine {
                RegexEngine::RUST_REGEX => {
                    let lock_cache = || REGEX_CACHE.lock_py_attached(py).unwrap_or_else(PoisonError::into_inner);
                    let re_pattern = if let Some(re_pattern) = lock_cache().get(&pattern_str) {
                        re_pattern.clone()
                    } else {
                        // compile the pattern outside of the lock so that other threads can use the cache while we compile
                        let pattern = Regex::new(&pattern_str).map_err(|e| py_schema_error_type!("{}", e))?;
                        // use get_or_insert to allow the possibility another thread raced to compile the same pattern
                        lock_cache().get_or_insert(pattern_str.clone(), || pattern).clone()
                    };

                    RegexEngine::RustRegex(re_pattern)
                }
                RegexEngine::PYTHON_RE => RegexEngine::PythonRe(re_compile.call1((pattern,))?.into()),
                _ => return Err(py_schema_error_type!("Invalid regex engine: {engine}")),
            };

            Ok(Self {
                pattern: pattern_str,
                engine,
            })
        }
    }

    fn is_match(&self, py: Python<'_>, target: &str) -> PyResult<bool> {
        match &self.engine {
            RegexEngine::RustRegex(regex) => Ok(regex.is_match(target)),
            RegexEngine::PythonRe(py_regex) => {
                Ok(!py_regex.call_method1(py, intern!(py, "search"), (target,))?.is_none(py))
            }
        }
    }
}
