from __future__ import annotations

from pipecat.frames.frames import (
    Frame,
    InterruptionFrame,
    LLMFullResponseEndFrame,
    LLMFullResponseStartFrame,
    LLMTextFrame,
    OutputTransportMessageUrgentFrame,
    TTSSpeakFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor


class ProductTextStreamProcessor(FrameProcessor):
    """Mirrors LLM text frames as streaming protocol events."""

    def __init__(self) -> None:
        super().__init__()
        self._aggregation: list[str] = []
        self._turn_active = False

    async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:
        await super().process_frame(frame, direction)

        if isinstance(frame, LLMFullResponseStartFrame):
            await self._start_turn()
        elif isinstance(frame, LLMTextFrame):
            if frame.text:
                await self._delta(frame.text)
        elif isinstance(frame, LLMFullResponseEndFrame):
            await self._end_turn(interrupted=False)
        elif isinstance(frame, InterruptionFrame):
            await self._end_turn(interrupted=True)
        elif isinstance(frame, TTSSpeakFrame):
            text = frame.text or ""
            await self._start_turn()
            if text:
                await self._delta(text)
            await self._end_turn(interrupted=False)

        await self.push_frame(frame, direction)

    async def _start_turn(self) -> None:
        if self._turn_active:
            return
        self._turn_active = True
        self._aggregation = []
        await self._emit("response.text.started")

    async def _delta(self, text: str) -> None:
        if not self._turn_active:
            await self._start_turn()
        self._aggregation.append(text)
        await self._emit("response.text.delta", text=text)

    async def _end_turn(self, *, interrupted: bool) -> None:
        if not self._turn_active:
            return
        full_text = "".join(self._aggregation)
        self._turn_active = False
        self._aggregation = []
        await self._emit(
            "response.text.final",
            text=full_text,
            interrupted=interrupted,
        )

    async def _emit(self, event_type: str, **payload: object) -> None:
        await self.push_frame(
            OutputTransportMessageUrgentFrame(
                message={"type": event_type, **payload},
            ),
            FrameDirection.DOWNSTREAM,
        )
