"""Text-processing helpers used by the Session pipeline.

Combines sentence-boundary detection, markdown delimiter tracking, speech
energy heuristics, and TTS payload text normalisation in one internal
module.  Everything here is a pure function / dataclass — no I/O, no
provider dependencies, safe to import from tests.

Naming convention for this module:

- Names in ``__all__`` (``split_at_sentence_boundaries``,
  ``has_unclosed_markdown_delimiters``, ``markdown_open_state``) are the
  supported public surface, re-exported from ``easycat.session`` /
  consumed by ``_streaming``.
- The leading-underscore helpers (``_truncate_partial_text_to_boundary``,
  ``_text_for_estimation_timeline``, ``_cleanup_estimation_text``,
  ``_chunk_has_speech_energy``) are package-internal cross-module API: they
  are imported by sibling ``session/`` modules (``interruption``,
  ``_turn_runner``, ``_audio_router``) but are *not* part of the supported
  public surface, so they keep the underscore. The remaining underscore
  helpers (``_is_word_char``, ``_has_unclosed_single_emphasis``,
  ``_scan_markdown_link_or_image``, ``_has_unclosed_markdown_link_or_image``)
  are file-local.
"""

from __future__ import annotations

import re
import struct

import sentencesplit

__all__ = [
    "has_unclosed_markdown_delimiters",
    "markdown_open_state",
    "split_at_sentence_boundaries",
    "split_first_clause",
]

from easycat.audio_format import AudioChunk
from easycat.tts.input import TTSInput, strip_ssml_tags

# ── Sentence splitting ──────────────────────────────────────────────
#
# sentencesplit's ``segment_with_lookahead`` probes tiny suffixes to
# detect whether the final boundary could still shift once more streaming
# text arrives (e.g. "GPT 3." might become "GPT 3.5").
_SENTENCE_SEGMENTER = sentencesplit.Segmenter(language="en", clean=False)


def split_at_sentence_boundaries(text: str) -> tuple[str, str]:
    """Split text at the last stable sentence boundary.

    Returns ``(ready_text, remaining_buffer)``.  ``ready_text`` contains
    complete sentences safe to send to TTS; ``remaining_buffer`` holds
    trailing text whose sentence boundary could still shift as more
    streaming text arrives.  Callers must flush the buffer when the
    stream ends.
    """
    if not text:
        return "", ""
    if not text.strip():
        return "", text

    result = _SENTENCE_SEGMENTER.segment_with_lookahead(text)
    if not result.segments:
        return "", text
    if not result.should_wait_for_more:
        return text, ""
    if len(result.segments) == 1:
        return "", text
    last = result.segments[-1]
    # ``char_span=True`` yields TextSpan objects with offsets; the plain
    # string variant of the union is narrowed via suffix length.
    last_start = len(text) - len(last) if isinstance(last, str) else last.start
    return text[:last_start], text[last_start:]


# Minimum length (in stripped characters) the first clause must reach before
# it is shipped on its own; below this we wait for the full sentence so a
# clipped opener like "Sure," is never sent to TTS by itself.
_FIRST_CLAUSE_MIN_CHARS = 12

# If a model streams a long opener without punctuation, cap how much text the
# first TTS payload waits for.  The split still happens only at a word boundary
# and only after a useful phrase has accumulated; later payloads continue to
# use full-sentence granularity.
_FIRST_PHRASE_TARGET_CHARS = 48
_FIRST_PHRASE_MIN_CHARS = 24

# Clause boundaries used only for the first payload of a turn: the sentence
# terminators ``split_at_sentence_boundaries`` honours plus mid-sentence
# punctuation (comma/semicolon/colon) that marks a natural early pause.
_FIRST_CLAUSE_BOUNDARY_CHARS = ".!?。！？．,;:"
_URL_SCHEMES_HELD_FOR_LOOKAHEAD = frozenset({"ftp", "ftps", "http", "https", "ws", "wss"})
_URL_LEADING_WRAPPERS = "([{<\"'`"


def split_first_clause(text: str) -> tuple[str, str]:
    """Split text at the first natural clause boundary.

    Returns ``(ready_text, remaining_buffer)``.  Used only for the *first*
    TTS payload of a turn to cut time-to-first-audio: emit at the first
    clause boundary (``,``/``;``/``:`` plus the sentence terminators
    :func:`split_at_sentence_boundaries` honours) instead of waiting for a
    full sentence.

    ``ready_text`` is non-empty only when the clause has at least
    ``_FIRST_CLAUSE_MIN_CHARS`` stripped characters, so a clipped opener
    like ``"Sure,"`` is never shipped on its own; in that case ``("",
    text)`` is returned and the caller falls back to full-sentence
    splitting.
    """
    if not text:
        return "", ""
    if not text.strip():
        return "", text

    for i, ch in enumerate(text):
        if ch not in _FIRST_CLAUSE_BOUNDARY_CHARS:
            continue
        if _is_url_separator(text, i):
            continue
        if _is_numeric_separator(text, i):
            continue
        # Include any trailing whitespace so the remaining buffer starts at
        # the next clause's first non-space character.
        end = i + 1
        while end < len(text) and text[end].isspace():
            end += 1
        ready = text[:end]
        if len(ready.strip()) >= _FIRST_CLAUSE_MIN_CHARS:
            return ready, text[end:]
        # Too short to ship on its own; keep scanning for a later boundary.

    return "", text


def _split_first_phrase(text: str) -> tuple[str, str]:
    """Bound the first TTS payload when streamed text has no punctuation.

    Once at least ``_FIRST_PHRASE_TARGET_CHARS`` are buffered, split at the
    latest whitespace at or before that target.  Requiring a boundary after
    ``_FIRST_PHRASE_MIN_CHARS`` avoids shipping a tiny prefix before a long
    URL, identifier, or unbroken word.  The whitespace stays with the ready
    prefix so concatenating the returned pair reproduces the input exactly.
    """
    if len(text) < _FIRST_PHRASE_TARGET_CHARS:
        return "", text

    upper = min(len(text), _FIRST_PHRASE_TARGET_CHARS + 1)
    split_at = -1
    for index in range(upper - 1, _FIRST_PHRASE_MIN_CHARS - 1, -1):
        if text[index].isspace():
            split_at = index + 1
            break
    if split_at < 0:
        return "", text

    while split_at < len(text) and text[split_at].isspace():
        split_at += 1
    return text[:split_at], text[split_at:]


def _is_url_separator(text: str, index: int) -> bool:
    start = index - 1
    while start >= 0 and not text[start].isspace():
        start -= 1
    token_prefix = text[start + 1 : index].lstrip(_URL_LEADING_WRAPPERS)
    if text[index] == ":" and text[index + 1 : index + 3] == "//":
        scheme = token_prefix
        return _is_url_scheme(scheme)
    if text[index] == ":" and index + 1 == len(text):
        scheme = token_prefix.lower()
        return scheme in _URL_SCHEMES_HELD_FOR_LOOKAHEAD

    scheme, sep, _ = token_prefix.partition("://")
    if not sep or not _is_url_scheme(scheme):
        return False
    return index + 1 < len(text) and not text[index + 1].isspace()


def _is_url_scheme(scheme: str) -> bool:
    return (
        bool(scheme) and scheme[0].isalpha() and all(ch.isalnum() or ch in "+-." for ch in scheme)
    )


def _is_numeric_separator(text: str, index: int) -> bool:
    if text[index] not in ".,．，:":
        return False
    if text[index] == ":" and index > 0 and text[index - 1].isdigit():
        return index + 1 == len(text) or text[index + 1].isdigit()
    if text[index] in ".．" and index > 0 and text[index - 1].isdigit():  # noqa: SIM102 nested branches preserve decision context
        # At the end of a streaming buffer, a numeric period may still become
        # a decimal separator once the next delta arrives (for example "3."
        # followed by "5").
        if index + 1 == len(text):
            return True
    return (
        index > 0
        and index + 1 < len(text)
        and text[index - 1].isdigit()
        and text[index + 1].isdigit()
    )


def _is_word_char(ch: str) -> bool:
    return ch.isalnum() or ch == "_"


def _truncate_partial_text_to_boundary(text: str, chars: int) -> str:
    """Trim a partial text estimate to a safer boundary.

    If the proportional cut lands in the middle of a word, trim back to
    the nearest non-word boundary so interruption context looks less
    noisy.  ``_is_word_char`` treats every alphanumeric character
    (including CJK) as word, so this is effectively a no-op for
    languages without whitespace word boundaries.
    """
    if chars <= 0:
        return ""
    if chars >= len(text):
        return text

    prefix = text[:chars]
    next_char = text[chars]

    # If the cut already lands on a boundary, keep the proportional
    # estimate as-is.
    if (not _is_word_char(prefix[-1])) or (not _is_word_char(next_char)):
        return prefix

    # Otherwise trim back to the nearest safe boundary in the prefix.
    for i in range(len(prefix) - 1, -1, -1):
        if not _is_word_char(prefix[i]):
            return prefix[: i + 1]

    # Single long token with no internal boundary; keep the proportional
    # cut.
    return prefix


# ── Speech energy heuristic ─────────────────────────────────────────


def _chunk_has_speech_energy(chunk: AudioChunk, *, threshold: int = 500) -> bool:
    """Heuristic speech gate for STT-driven turns when VAD is disabled.

    Computes the peak absolute PCM sample value for mono 16-bit chunks
    and compares it to ``threshold``.  Filters continuous silent /
    background frames (e.g. telephony keepalive silence) so they don't
    spuriously start turns.

    Runs on the audio ingress loop for every received frame while IDLE,
    so the peak is computed with a single batch decode (numpy when
    available, else ``struct.unpack``) rather than a per-sample Python
    loop, matching the rest of the audio pipeline in ``_audio_utils``.
    """
    if chunk.format.sample_width != 2:
        return bool(chunk.data)

    # Drop an odd trailing byte that would otherwise split a 16-bit sample.
    data = chunk.data
    data = data[: len(data) // 2 * 2]
    if not data:
        return False

    try:
        import numpy as np

        # Widen to int32 before abs so abs(-32768) does not overflow int16.
        samples = np.frombuffer(data, dtype="<i2").astype(np.int32)
        return bool(np.abs(samples).max() >= threshold)
    except ImportError:
        decoded = struct.unpack(f"<{len(data) // 2}h", data)
        return max(abs(s) for s in decoded) >= threshold


# ── Markdown delimiter tracking ─────────────────────────────────────


def _has_unclosed_single_emphasis(text: str, delimiter: str) -> bool:
    """Detect unclosed single-char emphasis delimiters (* / _) in text."""
    open_count = 0
    i = 0
    length = len(text)
    while i < length:
        ch = text[i]
        if ch == "\\":
            i += 2  # Skip escaped character (if any).
            continue
        if ch != delimiter:
            i += 1
            continue

        prev_char = text[i - 1] if i > 0 else ""
        next_char = text[i + 1] if i + 1 < length else ""

        # Ignore repeated runs (** / __ / *** / ___); those are handled
        # elsewhere.
        if prev_char == delimiter or next_char == delimiter:
            i += 1
            continue

        is_open = (not _is_word_char(prev_char)) and bool(next_char) and not next_char.isspace()
        is_close = bool(prev_char) and not prev_char.isspace() and (not _is_word_char(next_char))

        if is_close and open_count > 0:
            open_count -= 1
        elif is_open:
            open_count += 1
        i += 1

    return open_count > 0


def _scan_markdown_link_or_image(text: str) -> tuple[bool, bool]:
    """Scan ``text`` for incomplete markdown link/image spans.

    Returns ``(is_open, awaiting_destination)``:

    - ``is_open`` is ``True`` when a link/image span is still incomplete
      (an unbalanced label ``[``, a closed label ``[label]`` whose
      destination has not yet been observed, or an unbalanced destination
      ``(``).
    - ``awaiting_destination`` is ``True`` only for the specific case of a
      trailing closed label ``[label]`` whose next non-space character has
      not yet arrived — i.e. the span is open *solely* because we cannot
      yet tell whether a ``(destination)`` follows.  This is the one open
      case that a non-``(`` continuation can disambiguate, so the streaming
      path treats it differently from genuinely-open emphasis/code spans.

    Intentionally conservative for streaming safety: any trailing
    ``[label]`` without a resolved destination is treated as still-open
    until a non-link continuation is observed.
    """
    label_depth = 0
    awaiting_destination = False
    destination_depth = 0
    i = 0
    length = len(text)

    while i < length:
        ch = text[i]

        if ch == "\\":
            i += 2
            continue

        if destination_depth > 0:
            if ch == "(":
                destination_depth += 1
            elif ch == ")":
                destination_depth -= 1
            i += 1
            continue

        if label_depth > 0:
            if ch == "[":
                label_depth += 1
            elif ch == "]":
                label_depth -= 1
                if label_depth == 0:
                    awaiting_destination = True
            i += 1
            continue

        if awaiting_destination:
            if ch.isspace():
                i += 1
                continue
            if ch == "(":
                awaiting_destination = False
                destination_depth = 1
                i += 1
                continue
            # Closed bracket not followed by destination: not a markdown
            # link.
            awaiting_destination = False
            continue

        if ch == "[":
            label_depth = 1
        i += 1

    is_open = label_depth > 0 or awaiting_destination or destination_depth > 0
    return is_open, awaiting_destination


def _has_unclosed_markdown_link_or_image(text: str) -> bool:
    """Detect incomplete markdown link/image spans in ``text``."""
    return _scan_markdown_link_or_image(text)[0]


def markdown_open_state(text: str) -> tuple[bool, bool]:
    """Best-effort markdown openness check for a rolling streaming buffer.

    Returns ``(is_open, awaiting_link_destination)``:

    - ``is_open`` is ``True`` when any markdown span (fenced/inline code,
      ``**``/``__``/``~~`` runs, single ``*``/``_`` emphasis, or a
      link/image) is still unfinished, so later deltas could rewrite
      already-buffered text and emission must be deferred.
    - ``awaiting_link_destination`` is ``True`` only when the *sole* reason
      the buffer is open is a trailing closed label ``[label]`` whose
      ``(destination)`` (or lack thereof) has not yet arrived.  In that
      case any non-space, non-``(`` continuation disambiguates the span,
      so the streaming path can recheck eagerly rather than waiting for a
      markdown-closer character.

    The streaming path defers sentence emission while markdown delimiters
    are still open so later deltas cannot rewrite already-emitted text.
    """
    fenced_count = text.count("```")
    if fenced_count % 2 == 1:
        return True, False

    # Remove fenced blocks so inline delimiter counts are not distorted.
    normalized = re.sub(r"```[\s\S]*?```", "", text)

    # Inline backticks only (exclude fenced markers already handled
    # above).
    inline_tick_count = normalized.count("`")
    if inline_tick_count % 2 == 1:
        return True, False

    # Remove closed inline-code spans so markdown chars inside code do
    # not affect emphasis/link-state tracking.
    normalized = re.sub(r"`[^`]*`", "", normalized)

    for delimiter in ("**", "__", "~~"):
        if normalized.count(delimiter) % 2 == 1:
            return True, False

    link_open, awaiting_destination = _scan_markdown_link_or_image(normalized)

    emphasis_open = _has_unclosed_single_emphasis(
        normalized, "*"
    ) or _has_unclosed_single_emphasis(normalized, "_")

    is_open = link_open or emphasis_open
    # The awaiting-destination shortcut only applies when nothing else
    # holds the window open; an unclosed emphasis span still needs a
    # genuine closer.
    return is_open, awaiting_destination and not emphasis_open


def has_unclosed_markdown_delimiters(text: str) -> bool:
    """Best-effort check for unfinished markdown spans in a rolling buffer.

    The streaming path defers sentence emission while markdown delimiters
    are still open so later deltas cannot rewrite already-emitted text.
    """
    return markdown_open_state(text)[0]


# ── TTS payload text normalisation for interruption estimation ──────

_PAUSE_MARKER = ""
_PAUSE_CHARS_PER_SECOND = 14.0


def _text_for_estimation_timeline(payload: TTSInput) -> str:
    """Return text used for interruption timeline estimation.

    For SSML payloads, explicit ``<break .../>`` pauses are expanded
    into synthetic marker characters so byte→text interpolation accounts
    for non-spoken silence regions.
    """
    if payload.format != "ssml":
        return payload.text

    def _break_repl(match: re.Match[str]) -> str:
        attrs = match.group(1)
        time_match = re.search(
            r"""time\s*=\s*(['"])\s*((?:\d+(?:\.\d+)?)|(?:\.\d+))\s*(ms|s)\s*\1""",
            attrs,
            flags=re.IGNORECASE,
        )
        ms = 0.0
        if time_match:
            amount = float(time_match.group(2))
            unit = time_match.group(3).lower()
            ms = amount if unit == "ms" else amount * 1000.0
        count = max(1, round((ms / 1000.0) * _PAUSE_CHARS_PER_SECOND)) if ms > 0 else 1
        return _PAUSE_MARKER * count

    with_markers = re.sub(r"<break\b([^>]*)/>", _break_repl, payload.text, flags=re.IGNORECASE)
    return strip_ssml_tags(with_markers)


def _cleanup_estimation_text(text: str) -> str:
    """Remove synthetic pause markers from estimated spoken text."""
    return text.replace(_PAUSE_MARKER, "")
