from __future__ import annotations

import re
from dataclasses import dataclass
from urllib.parse import unquote

from errors import IncrementalError
from metadata import TableName
from snapshot.schema import quote_object_path_segment

SCHEMA_RE = re.compile(r"^schema_(?P<table_version>\d+)_(?P<checksum>\d+)\.json$")
INDEX_RE = re.compile(r"^CDC(?:_(?P<dispatcher_id>[^.]+))?\.index$")
DATA_RE = re.compile(r"^CDC(?P<file_index>\d+)\.csv$")
DISPATCHER_DATA_RE = re.compile(r"^CDC_(?P<dispatcher_id>.+)_(?P<file_index>\d+)\.csv$")
DATE_RE = re.compile(r"^\d{4}(?:-\d{2})?(?:-\d{2})?$")


@dataclass(frozen=True)
class SchemaPath:
    key: str
    source_schema: str
    source_table: str
    table_version: int
    checksum: int

    @property
    def table_name(self) -> TableName:
        return TableName(self.source_schema, self.source_table)


@dataclass(frozen=True)
class IndexPath:
    key: str
    source_schema: str
    source_table: str
    table_version: int
    partition_id: int
    date: str
    dispatcher_id: str
    data_dir: str
    latest_file_index: int
    file_width: int

    @property
    def table_name(self) -> TableName:
        return TableName(self.source_schema, self.source_table)

    @property
    def quoted(self) -> str:
        return self.table_name.quoted

    @property
    def cursor_key(self) -> str:
        return self.key

    @property
    def sort_key(self) -> tuple[int, int, str, str, str]:
        return (
            self.table_version,
            self.partition_id,
            self.date,
            self.source_schema,
            self.source_table,
        )

    def data_file_key(self, file_index: int) -> str:
        return f"{self.data_dir}/{data_file_name(file_index, self.file_width)}"


def is_index_key(key: str) -> bool:
    filename = key.rsplit("/", 1)[-1]
    return INDEX_RE.match(filename) is not None


def data_file_name(file_index: int, file_width: int) -> str:
    return f"CDC{file_index:0{file_width}d}.csv"


def _strip_increment_prefix(key: str, increment_prefix: str) -> str:
    prefix = increment_prefix.strip("/")
    key = key.strip("/")
    if prefix and key.startswith(prefix + "/"):
        return key[len(prefix) + 1 :]
    return key


def _decode_object_path_segment(segment: str, key: str) -> str:
    decoded = unquote(segment)
    if not decoded or decoded in {".", ".."} or "/" in decoded or "\\" in decoded:
        raise IncrementalError(f"invalid incremental object path segment {segment!r} in {key}")
    if quote_object_path_segment(decoded) != segment:
        raise IncrementalError(f"incremental object path segment is not normalized: {segment!r} in {key}")
    return decoded


def parse_schema_path(key: str, increment_prefix: str) -> SchemaPath | None:
    relative = _strip_increment_prefix(key, increment_prefix)
    parts = relative.split("/")
    if len(parts) == 3 and parts[1] == "meta":
        source_table = ""
        filename = parts[2]
    elif len(parts) == 4 and parts[2] == "meta":
        source_table = _decode_object_path_segment(parts[1], key)
        filename = parts[3]
    else:
        return None
    match = SCHEMA_RE.match(filename)
    if not match:
        return None
    return SchemaPath(
        key=key,
        source_schema=_decode_object_path_segment(parts[0], key),
        source_table=source_table,
        table_version=int(match.group("table_version")),
        checksum=int(match.group("checksum")),
    )


def parse_index_content(content: str) -> tuple[int, int, str]:
    filename = content.strip().splitlines()[0].strip() if content.strip() else ""
    match = DATA_RE.match(filename)
    if match:
        return int(match.group("file_index")), len(match.group("file_index")), ""
    match = DISPATCHER_DATA_RE.match(filename)
    if match:
        return (
            int(match.group("file_index")),
            len(match.group("file_index")),
            match.group("dispatcher_id"),
        )
    raise IncrementalError(f"invalid incremental index content: {content!r}")


def _parse_date_and_partition(path_parts: list[str]) -> tuple[int, str]:
    if not path_parts:
        return 0, ""
    date = ""
    partition_id = 0
    if DATE_RE.match(path_parts[-1]):
        date = path_parts[-1]
        path_parts = path_parts[:-1]
    if path_parts:
        try:
            partition_id = int(path_parts[-1])
        except ValueError as exc:
            raise IncrementalError(f"invalid partition path segment: {path_parts[-1]!r}") from exc
    return partition_id, date


def parse_index_path(key: str, increment_prefix: str, content: str) -> IndexPath | None:
    relative = _strip_increment_prefix(key, increment_prefix)
    parts = relative.split("/")
    if len(parts) < 5 or parts[-2] != "meta":
        return None
    file_match = INDEX_RE.match(parts[-1])
    if not file_match:
        return None
    dispatcher_id = file_match.group("dispatcher_id") or ""
    latest_file_index, file_width, dispatcher_from_data = parse_index_content(content)
    if dispatcher_from_data and dispatcher_id and dispatcher_id != dispatcher_from_data:
        raise IncrementalError(f"dispatcher ID mismatch between index path and content: {key}")
    dispatcher_id = dispatcher_id or dispatcher_from_data
    if dispatcher_id:
        raise IncrementalError("v1 does not support table-across-nodes dispatcher index files")

    try:
        table_version = int(parts[2])
    except ValueError as exc:
        raise IncrementalError(f"invalid tableVersion in index path: {key}") from exc

    partition_id, date = _parse_date_and_partition(parts[3:-2])
    data_dir = "/".join(key.strip("/").split("/")[:-2])
    return IndexPath(
        key=key,
        source_schema=_decode_object_path_segment(parts[0], key),
        source_table=_decode_object_path_segment(parts[1], key),
        table_version=table_version,
        partition_id=partition_id,
        date=date,
        dispatcher_id=dispatcher_id,
        data_dir=data_dir,
        latest_file_index=latest_file_index,
        file_width=file_width,
    )


__all__ = [
    "IndexPath",
    "SchemaPath",
    "data_file_name",
    "is_index_key",
    "parse_index_content",
    "parse_index_path",
    "parse_schema_path",
]