import asyncio
import itertools
import logging
import math
import sys
from abc import ABC, abstractmethod
from collections.abc import Callable
from pathlib import Path
from typing import Any, Generator, Generic, Iterable, TextIO, TypeVar

ReaderT = TypeVar('ReaderT')
WriterT = TypeVar('WriterT')

logger = logging.getLogger('users_cleanup')


class BaseReader(Generic[ReaderT], ABC):
    def __init__(self, *, path: Path, row_filter: Callable[[ReaderT], bool] | None = None) -> None:
        self.path = path
        self.row_filter = row_filter
        self._fp: TextIO | None = None
        self._lines: int = 0

    def __enter__(self) -> 'BaseReader[ReaderT]':
        self._fp = self.path.open('r')
        self._lines = sum(1 for _ in self._fp)
        self._fp.seek(0)
        return self

    def __exit__(self, *args: Any, **kwargs: Any) -> None:
        if self._fp is not None:
            self._fp.close()
        self._fp = None

    def __len__(self) -> int:
        """Number of lines in the file"""
        return self._lines

    @abstractmethod
    def __iter__(self) -> Generator[Any, None, None]:
        pass

    @abstractmethod
    def _get_row(self) -> Generator[ReaderT, None, None]:
        """Yields items from the file."""
        ...


class BaseRowReader(BaseReader[ReaderT], ABC):
    def __iter__(self) -> Generator[tuple[int, ReaderT], None, None]:
        for index, row in enumerate(self._get_row()):
            logger.info(f'Reading row {index + 1}/{self._lines}')
            if self.row_filter and not self.row_filter(row):
                continue
            yield index, row


class BaseBatchReader(BaseReader[ReaderT], ABC):
    def __init__(self, *, batch_size: int, path: Path, row_filter: Callable[[ReaderT], bool] | None = None) -> None:
        self.batch_size = batch_size
        super().__init__(path=path, row_filter=row_filter)

    def __iter__(self) -> Generator[tuple[int, list[ReaderT]], None, None]:
        iterator: Iterable[ReaderT] = iter(self._get_row())
        if self.row_filter:
            iterator = filter(self.row_filter, iterator)

        batches = math.ceil(self._lines / self.batch_size)

        index = 0
        while batch := list(itertools.islice(iterator, self.batch_size)):
            logger.info(f'Reading batch {index + 1}/{batches}')
            yield index, batch
            index += 1


class BaseWriter(Generic[WriterT], ABC):
    def __init__(self, *, path: Path, queue: asyncio.Queue[WriterT]) -> None:
        self.path = path
        self.queue = queue
        self._fp: TextIO | None = None

    def __enter__(self) -> 'BaseWriter[WriterT]':
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self._fp = self.path.open('w')
        return self

    def __exit__(self, *args: Any, **kwargs: Any) -> None:
        if self._fp is not None:
            self._fp.close()
        self._fp = None

    @abstractmethod
    def _write_record(self, record: WriterT) -> None:
        """Writes the given message to the queue."""

    async def wait_for_records(self) -> None:
        """Consumes the queue and writes to the path."""
        logger.info(f'Writing results to: {self.path}')
        try:
            while True:
                record = await self.queue.get()
                self._write_record(record)
                self.queue.task_done()
        except asyncio.QueueShutDown:
            logger.debug('Result writer finished.')


class BaseAsyncFileProcessor(Generic[ReaderT, WriterT], ABC):
    """
    Base async processor using Strategy Pattern.
    Handles concurrency and flow control.
    Delegates IO to strategies.
    """

    def __init__(self, *, fi: Path, fo: Path, concurrency: int) -> None:
        self.fi = fi
        self.fo = fo

        self.concurrency = concurrency
        self.max_pending_tasks = concurrency * 2
        self.semaphore = asyncio.Semaphore(self.concurrency)
        self.pending_tasks: set[asyncio.Task[None]] = set()
        self.result_queue: asyncio.Queue[WriterT] = asyncio.Queue()

    @abstractmethod
    def get_reader(self) -> BaseReader[ReaderT]:
        """Returns the reader strategy."""

    @abstractmethod
    def get_writer(self) -> BaseWriter[WriterT]:
        """Returns the writer strategy."""

    @abstractmethod
    async def process_row(self, index: int, row: ReaderT) -> WriterT:
        """
        Business Logic is the only thing left abstract!
        """
        pass

    async def pre_process(self) -> None:
        pass

    async def post_process(self) -> None:
        pass

    async def _process_row(self, index: int, row: ReaderT) -> None:
        async with self.semaphore:
            result = await self.process_row(index, row)
            self.result_queue.put_nowait(result)

    async def process(self) -> None:
        with self.get_writer() as writer, self.get_reader() as reader:
            writer_task = asyncio.create_task(writer.wait_for_records())

            logger.info(f'Processing {self.fi.name}...')
            await self.pre_process()

            try:
                for index, value in reader:
                    task = asyncio.create_task(self._process_row(index, value))
                    self.pending_tasks.add(task)
                    task.add_done_callback(self.pending_tasks.discard)

                    if len(self.pending_tasks) >= self.max_pending_tasks:
                        await asyncio.wait(self.pending_tasks, return_when=asyncio.FIRST_COMPLETED)

                if self.pending_tasks:
                    await asyncio.wait(self.pending_tasks)

                await self.post_process()

                self.result_queue.shutdown()
                await writer_task
                logger.info('All processing complete')

            except Exception as e:
                logger.exception(f'Critical execution error: {e}')
                self.result_queue.shutdown(immediate=True)
                await writer_task
                sys.exit(1)