import sys
from contextlib import contextmanager
from pathlib import Path
from typing import Annotated, Iterator, Optional, Tuple

from pydantic import AfterValidator, BaseModel
from pydantic_settings import BaseSettings, SettingsConfigDict


def _validate_page_range(v: Tuple[int, int]) -> Tuple[int, int]:
    if v[0] < 1 or v[1] < v[0]:
        raise ValueError(
            "Invalid page range: start must be ≥ 1 and end must be ≥ start."
        )
    return v


PageRange = Annotated[Tuple[int, int], AfterValidator(_validate_page_range)]

DEFAULT_PAGE_RANGE: PageRange = (1, sys.maxsize)


class DocumentLimits(BaseModel):
    max_num_pages: int = sys.maxsize
    max_file_size: int = sys.maxsize
    page_range: PageRange = DEFAULT_PAGE_RANGE


class BatchConcurrencySettings(BaseModel):
    doc_batch_size: int = 1  # Number of documents processed in one batch. Should be >= doc_batch_concurrency
    doc_batch_concurrency: int = 1  # Number of parallel threads processing documents. Warning: Experimental! No benefit expected without free-threaded python.
    page_batch_size: int = 4  # Number of pages processed in one batch.
    page_batch_concurrency: int = 1  # Currently unused.
    elements_batch_size: int = (
        16  # Number of elements processed in one batch, in enrichment models.
    )

    # To force models into single core: export OMP_NUM_THREADS=1


class DebugSettings(BaseModel):
    visualize_cells: bool = False
    visualize_ocr: bool = False
    visualize_layout: bool = False
    visualize_raw_layout: bool = False
    visualize_tables: bool = False

    profile_pipeline_timings: bool = False

    # Path used to output debug information.
    debug_output_path: str = str(Path.cwd() / "debug")


class InferenceSettings(BaseModel):
    compile_torch_models: bool = True


class AppSettings(BaseSettings):
    model_config = SettingsConfigDict(
        env_prefix="DOCLING_", env_nested_delimiter="_", env_nested_max_split=1
    )

    perf: BatchConcurrencySettings = BatchConcurrencySettings()
    debug: DebugSettings = DebugSettings()
    inference: InferenceSettings = InferenceSettings()

    cache_dir: Path = Path.home() / ".cache" / "docling"
    artifacts_path: Optional[Path] = None


settings = AppSettings()


def defaults() -> AppSettings:
    """Return a fresh settings instance populated from the current environment."""
    return AppSettings()


@contextmanager
def scoped(
    *,
    perf: BatchConcurrencySettings | None = None,
    debug: DebugSettings | None = None,
    inference: InferenceSettings | None = None,
) -> Iterator[AppSettings]:
    """Temporarily override selected settings and restore them on exit."""
    saved = {
        "perf": settings.perf.model_copy(deep=True),
        "debug": settings.debug.model_copy(deep=True),
        "inference": settings.inference.model_copy(deep=True),
    }

    try:
        if perf is not None:
            settings.perf = perf
        if debug is not None:
            settings.debug = debug
        if inference is not None:
            settings.inference = inference
        yield settings
    finally:
        settings.perf = saved["perf"].model_copy(deep=True)
        settings.debug = saved["debug"].model_copy(deep=True)
        settings.inference = saved["inference"].model_copy(deep=True)


def default_compile_model() -> bool:
    return settings.inference.compile_torch_models
