from __future__ import annotations

import json
from collections.abc import Mapping
from typing import Any, Literal, cast
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit

from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator

from errors import ConfigurationError

CONFIG_CHECKPOINT_INTERVAL_ROWS = "checkpoint_interval_rows"
CONFIG_CHECKPOINT_INTERVAL_SECONDS = "checkpoint_interval_seconds"
CONFIG_CSV_BINARY_ENCODING = "csv_binary_encoding"
CONFIG_CSV_BINARY_ENCODING_METHOD = "csv_binary_encoding_method"
CONFIG_CSV_BINARY_OUTPUT_DIALECT = "csv_binary_output_dialect"
CONFIG_CSV_DELIMITER = "csv_delimiter"
CONFIG_CSV_LINE_TERMINATOR = "csv_line_terminator"
CONFIG_CSV_NO_HEADER = "csv_no_header"
CONFIG_CSV_NULL_VALUE = "csv_null_value"
CONFIG_CSV_OUTPUT_OLD_VALUE = "csv_output_old_value"
CONFIG_CSV_QUOTE = "csv_quote"
CONFIG_CSV_SEPARATOR = "csv_separator"
CONFIG_ENABLE_DIAGNOSTIC_COLUMNS = "enable_diagnostic_columns"
CONFIG_ENABLE_TABLE_ACROSS_NODES = "enable_table_across_nodes"
CONFIG_ESCAPE_BACKSLASH = "escape_backslash"
CONFIG_INCREMENTAL_PREFIX = "incremental_prefix"
CONFIG_INCREMENTAL_BLOCK = "incremental"
CONFIG_OPERATION_MODE = "operation_mode"
CONFIG_S3_AK = "s3_ak"
CONFIG_S3_REGION = "s3_region"
CONFIG_S3_SK = "s3_sk"
CONFIG_S3_CSV_READ_MODE = "s3_csv_read_mode"
CONFIG_S3_URL = "s3_url"
CONFIG_SNAPSHOT_PREFIX = "snapshot_prefix"
CONFIG_SNAPSHOT_BLOCK = "snapshot"
CONFIG_SNAPSHOT_REQUIRED = "snapshot_required"
CONFIG_SYNC_TASK_ID = "sync_task_id"
CONFIG_TRUNCATE_BEFORE_SNAPSHOT = "truncate_before_snapshot"
OPERATION_MODE_BATCHED_STREAM = "batched_stream"
OPERATION_MODE_STANDARD = "standard"
S3_CSV_READ_MODE_BUFFERED = "buffered"
S3_CSV_READ_MODE_STREAMING = "streaming"

type OperationMode = Literal["batched_stream", "standard"]
type S3CSVReadMode = Literal["streaming", "buffered"]


def _coerce_bool(value: Any, *, default: bool = False) -> bool:
    if value is None:
        return default
    if isinstance(value, bool):
        return value
    text = str(value).strip().lower()
    if text in {"1", "true", "yes", "y", "on"}:
        return True
    if text in {"0", "false", "no", "n", "off"}:
        return False
    raise ConfigurationError(f"invalid boolean value: {value!r}")


def _coerce_true_false(value: Any, *, default: bool, path: str) -> bool:
    if value is None or value == "":
        return default
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        text = value.strip().lower()
        if text == "true":
            return True
        if text == "false":
            return False
    raise ConfigurationError(f"{path} must be true or false")


def _coerce_int(value: Any, *, default: int, minimum: int = 1) -> int:
    if value is None or value == "":
        return default
    try:
        result = int(value)
    except (TypeError, ValueError) as exc:
        raise ConfigurationError(f"invalid integer value: {value!r}") from exc
    if result < minimum:
        raise ConfigurationError(f"integer value must be >= {minimum}: {value!r}")
    return result


def _clean_prefix(value: str | None) -> str:
    if not value:
        return ""
    prefix = str(value).strip().strip("/")
    if not prefix:
        return ""
    parts = prefix.split("/")
    if any(part in {"", ".", ".."} for part in parts):
        raise ConfigurationError(f"prefix must be a normalized relative path: {value!r}")
    return prefix


def _validate_default_value(value: Any, expected: str, path: str) -> None:
    if value not in (None, "") and str(value) != expected:
        raise ConfigurationError(f"{path} must be {expected!r}")


def _validate_supported_csv_block(
    block: dict[str, Any],
    *,
    block_name: str,
    delimiter_key: str | None,
    delimiter_expected: str,
    separator_key: str | None,
    quote_key: str | None,
) -> None:
    if delimiter_key is not None:
        _validate_default_value(block.get(delimiter_key), delimiter_expected, f"{block_name}.{delimiter_key}")
    if separator_key is not None:
        _validate_default_value(block.get(separator_key), ",", f"{block_name}.{separator_key}")
    if quote_key is not None:
        _validate_default_value(block.get(quote_key), '"', f"{block_name}.{quote_key}")


def _validate_line_terminator(value: Any, path: str) -> None:
    if value in (None, ""):
        return
    if str(value) not in {"\n", "\r\n"}:
        raise ConfigurationError(f"{path} must be '\\n' or '\\r\\n'")


def _snapshot_binary_encoding_from_output_dialect(value: Any) -> tuple[str, str]:
    dialect = str(value or "default").strip().lower()
    if dialect in {"", "default"}:
        return "default", "utf8"
    if dialect in {"snowflake", "hex"}:
        return dialect, "hex"
    raise ConfigurationError("snapshot.csv_binary_output_dialect must be one of default/snowflake/hex")


def _incremental_csv_binary_encoding(block: dict[str, Any]) -> str:
    value = block.get(CONFIG_CSV_BINARY_ENCODING)
    if value in (None, ""):
        value = block.get(CONFIG_CSV_BINARY_ENCODING_METHOD)
    encoding = str(value or "base64").strip().lower()
    if encoding not in {"base64", "hex"}:
        raise ConfigurationError("incremental.csv_binary_encoding must be one of base64/hex")
    return encoding


class _ConnectorConfigurationInput(BaseModel):
    model_config = ConfigDict(extra="allow")

    sync_task_id: Any = None
    s3_url: Any = None
    snapshot_prefix: Any = None
    incremental_prefix: Any = None
    snapshot: dict[str, Any] = Field(default_factory=dict)
    incremental: dict[str, Any] = Field(default_factory=dict)

    @field_validator("snapshot", "incremental", mode="before")
    @classmethod
    def _parse_config_block(cls, value: Any, info: Any) -> dict[str, Any]:
        if value in (None, ""):
            return {}
        if isinstance(value, dict):
            return value
        if isinstance(value, str):
            if value.strip().lower() in {"true", "false"}:
                return {}
            try:
                parsed = json.loads(value)
            except json.JSONDecodeError as exc:
                raise ValueError(f"{info.field_name} must be a JSON object") from exc
            if isinstance(parsed, dict):
                return parsed
        raise ValueError(f"{info.field_name} must be an object")


class CheckpointConfig(BaseModel):
    model_config = ConfigDict(frozen=True)

    interval_rows: int = 50000
    interval_seconds: int = 600


class SnapshotConfig(BaseModel):
    model_config = ConfigDict(frozen=True)

    csv_null_value: str = r"\N"
    csv_binary_output_dialect: str = "default"
    binary_encoding: str = "utf8"
    escape_backslash: bool = False
    s3_csv_read_mode: S3CSVReadMode = S3_CSV_READ_MODE_STREAMING


class IncrementalConfig(BaseModel):
    model_config = ConfigDict(frozen=True)

    csv_null_value: str = r"\N"
    csv_binary_encoding: str = "base64"
    output_old_value: bool = False


class ConnectorConfig(BaseModel):
    model_config = ConfigDict(frozen=True)

    sync_task_id: str
    storage_uri: str
    snapshot_prefix: str
    incremental_prefix: str
    snapshot_required: bool = True
    snapshot: SnapshotConfig
    incremental: IncrementalConfig
    checkpoint: CheckpointConfig
    enable_diagnostic_columns: bool = False
    truncate_before_snapshot: bool = False
    operation_mode: OperationMode = OPERATION_MODE_BATCHED_STREAM

    @classmethod
    def from_dict(cls, configuration: object) -> ConnectorConfig:
        configuration = _validate_configuration_input(configuration).model_dump(mode="python")
        storage_uri = _resolve_storage_uri(configuration)
        if _coerce_bool(configuration.get(CONFIG_ENABLE_TABLE_ACROSS_NODES), default=False):
            raise ConfigurationError("v1 does not support enable_table_across_nodes=true")

        snapshot_block = _config_block(configuration, CONFIG_SNAPSHOT_BLOCK)
        incremental_block = _config_block(configuration, CONFIG_INCREMENTAL_BLOCK)

        snapshot_prefix = _required_prefix(
            configuration,
            CONFIG_SNAPSHOT_PREFIX,
        )
        incremental_prefix = _required_prefix(
            configuration,
            CONFIG_INCREMENTAL_PREFIX,
        )
        sync_task_id = _required_sync_task_id(configuration)
        _validate_csv_configuration(snapshot_block, incremental_block)

        return cls(
            sync_task_id=sync_task_id,
            storage_uri=storage_uri,
            snapshot_prefix=snapshot_prefix,
            incremental_prefix=incremental_prefix,
            snapshot_required=_coerce_bool(configuration.get(CONFIG_SNAPSHOT_REQUIRED), default=True),
            snapshot=_snapshot_config(snapshot_block),
            incremental=_incremental_config(incremental_block),
            checkpoint=_checkpoint_config(configuration),
            enable_diagnostic_columns=_coerce_bool(configuration.get(CONFIG_ENABLE_DIAGNOSTIC_COLUMNS), default=False),
            truncate_before_snapshot=_coerce_bool(configuration.get(CONFIG_TRUNCATE_BEFORE_SNAPSHOT), default=False),
            operation_mode=_operation_mode(configuration.get(CONFIG_OPERATION_MODE)),
        )


def _validate_configuration_input(configuration: object) -> _ConnectorConfigurationInput:
    if not isinstance(configuration, Mapping):
        raise ConfigurationError("configuration must be an object")
    configuration = dict(configuration.items())
    try:
        return _ConnectorConfigurationInput.model_validate(configuration)
    except ValidationError as exc:
        raise ConfigurationError(_pydantic_error_message(exc)) from exc


def _pydantic_error_message(exc: ValidationError) -> str:
    error = exc.errors()[0]
    ctx_error = error.get("ctx", {}).get("error")
    if ctx_error is not None:
        return str(ctx_error)
    return str(error.get("msg", exc))


def _config_block(configuration: dict[str, Any], key: str) -> dict[str, Any]:
    raw = configuration.get(key)
    if raw in (None, ""):
        return {}
    if not isinstance(raw, dict):
        if isinstance(raw, str):
            try:
                parsed = json.loads(raw)
            except json.JSONDecodeError as exc:
                raise ConfigurationError(f"{key} must be a JSON object") from exc
            if isinstance(parsed, dict):
                return parsed
        raise ConfigurationError(f"{key} must be an object")
    return raw


def _required_sync_task_id(configuration: dict[str, Any]) -> str:
    sync_task_id = str(configuration.get(CONFIG_SYNC_TASK_ID) or "").strip()
    if not sync_task_id:
        raise ConfigurationError("sync_task_id is required")
    return sync_task_id


def _validate_csv_configuration(snapshot_block: dict[str, Any], incremental_block: dict[str, Any]) -> None:
    _validate_supported_csv_block(
        snapshot_block,
        block_name=CONFIG_SNAPSHOT_BLOCK,
        delimiter_key=CONFIG_CSV_DELIMITER,
        delimiter_expected='"',
        separator_key=CONFIG_CSV_SEPARATOR,
        quote_key=None,
    )
    _validate_supported_csv_block(
        incremental_block,
        block_name=CONFIG_INCREMENTAL_BLOCK,
        delimiter_key=CONFIG_CSV_DELIMITER,
        delimiter_expected=",",
        separator_key=None,
        quote_key=CONFIG_CSV_QUOTE,
    )
    if CONFIG_CSV_NO_HEADER in snapshot_block:
        _coerce_bool(snapshot_block.get(CONFIG_CSV_NO_HEADER), default=False)
    if "csv_parser" in snapshot_block:
        raise ConfigurationError("snapshot.csv_parser is no longer supported")
    if CONFIG_CSV_LINE_TERMINATOR in snapshot_block:
        _validate_line_terminator(snapshot_block[CONFIG_CSV_LINE_TERMINATOR], "snapshot.csv_line_terminator")
    if CONFIG_CSV_LINE_TERMINATOR in incremental_block:
        _validate_line_terminator(incremental_block[CONFIG_CSV_LINE_TERMINATOR], "incremental.csv_line_terminator")


def _snapshot_config(snapshot_block: dict[str, Any]) -> SnapshotConfig:
    csv_binary_output_dialect, binary_encoding = _snapshot_binary_encoding_from_output_dialect(
        snapshot_block.get(CONFIG_CSV_BINARY_OUTPUT_DIALECT)
    )
    escape_backslash = _coerce_true_false(
        snapshot_block.get(CONFIG_ESCAPE_BACKSLASH),
        default=False,
        path="snapshot.escape_backslash",
    )
    return SnapshotConfig(
        csv_null_value=str(snapshot_block.get(CONFIG_CSV_NULL_VALUE) or r"\N"),
        csv_binary_output_dialect=csv_binary_output_dialect,
        binary_encoding=binary_encoding,
        escape_backslash=escape_backslash,
        s3_csv_read_mode=_s3_csv_read_mode(snapshot_block.get(CONFIG_S3_CSV_READ_MODE)),
    )


def _incremental_config(incremental_block: dict[str, Any]) -> IncrementalConfig:
    return IncrementalConfig(
        csv_null_value=str(incremental_block.get(CONFIG_CSV_NULL_VALUE) or r"\N"),
        csv_binary_encoding=_incremental_csv_binary_encoding(incremental_block),
        output_old_value=_coerce_bool(
            incremental_block.get(CONFIG_CSV_OUTPUT_OLD_VALUE),
            default=False,
        ),
    )


def _checkpoint_config(configuration: dict[str, Any]) -> CheckpointConfig:
    return CheckpointConfig(
        interval_rows=_coerce_int(configuration.get(CONFIG_CHECKPOINT_INTERVAL_ROWS), default=50000),
        interval_seconds=_coerce_int(configuration.get(CONFIG_CHECKPOINT_INTERVAL_SECONDS), default=600),
    )


def _operation_mode(value: Any) -> OperationMode:
    if value in (None, ""):
        return OPERATION_MODE_BATCHED_STREAM
    mode = str(value).strip().lower()
    if mode in {OPERATION_MODE_BATCHED_STREAM, OPERATION_MODE_STANDARD}:
        return cast(OperationMode, mode)
    raise ConfigurationError("operation_mode must be one of batched_stream/standard")


def _s3_csv_read_mode(value: Any) -> S3CSVReadMode:
    if value in (None, ""):
        return S3_CSV_READ_MODE_STREAMING
    mode = str(value).strip().lower()
    if mode in {S3_CSV_READ_MODE_STREAMING, S3_CSV_READ_MODE_BUFFERED}:
        return cast(S3CSVReadMode, mode)
    raise ConfigurationError("snapshot.s3_csv_read_mode must be one of streaming/buffered")


def _required_prefix(configuration: dict[str, Any], key: str) -> str:
    if key not in configuration or configuration.get(key) is None:
        raise ConfigurationError(f"{key} is required")
    return _clean_prefix(configuration.get(key))


def _resolve_storage_uri(cfg: dict[str, Any]) -> str:
    if cfg.get(CONFIG_S3_URL):
        return _merge_storage_uri_params(str(cfg[CONFIG_S3_URL]), cfg)
    raise ConfigurationError("s3_url is required")


def _merge_storage_uri_params(raw: str, cfg: dict[str, Any]) -> str:
    parsed = urlsplit(raw)
    if parsed.scheme.lower() != "s3":
        return raw

    params = parse_qsl(parsed.query, keep_blank_values=True)
    present = {key for key, _value in params}
    changed = False

    for uri_key, config_key in (
        ("region", CONFIG_S3_REGION),
        ("access-key", CONFIG_S3_AK),
        ("secret-access-key", CONFIG_S3_SK),
    ):
        value = cfg.get(config_key)
        if value not in (None, "") and uri_key not in present:
            params.append((uri_key, str(value)))
            present.add(uri_key)
            changed = True

    if not changed:
        return raw

    return urlunsplit(
        (
            parsed.scheme,
            parsed.netloc,
            parsed.path,
            urlencode(params),
            parsed.fragment,
        )
    )


__all__ = [
    "OPERATION_MODE_BATCHED_STREAM",
    "OPERATION_MODE_STANDARD",
    "S3_CSV_READ_MODE_BUFFERED",
    "S3_CSV_READ_MODE_STREAMING",
    "ConnectorConfig",
    "OperationMode",
    "S3CSVReadMode",
]