abogen/abogen/speaker_analysis.py

from __future__ import annotations

import re
from dataclasses import dataclass, field
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple

import unicodedata

_DIALOGUE_VERBS = (
    "said",
    "asked",
    "replied",
    "whispered",
    "shouted",
    "cried",
    "muttered",
    "answered",
    "hissed",
    "called",
    "added",
    "continued",
    "insisted",
    "remarked",
    "yelled",
    "breathed",
    "murmured",
    "exclaimed",
    "explained",
    "noted",
)

_VERB_PATTERN = "(?:" + "|".join(_DIALOGUE_VERBS) + ")"
_NAME_FRAGMENT = r"[A-ZÀ-ÖØ-Þ][\w'’\-]*"
_NAME_PATTERN = rf"{_NAME_FRAGMENT}(?:\s+{_NAME_FRAGMENT})*"

_COLON_PATTERN = re.compile(rf"^\s*({_NAME_PATTERN})\s*:\s*(.+)$")
_NAME_BEFORE_VERB = re.compile(rf"({_NAME_PATTERN})\s+{_VERB_PATTERN}\b", re.IGNORECASE)
_VERB_BEFORE_NAME = re.compile(rf"{_VERB_PATTERN}\s+({_NAME_PATTERN})", re.IGNORECASE)
_PRONOUN_PATTERN = re.compile(r"\b(?:he|she|they)\b", re.IGNORECASE)
_QUOTE_PATTERN = re.compile(r'["“”]([^"“”\\]*(?:\\.[^"“”\\]*)*)["”]')
_MALE_PRONOUN_PATTERN = re.compile(r"\b(?:he|him|his|himself)\b", re.IGNORECASE)
_FEMALE_PRONOUN_PATTERN = re.compile(r"\b(?:she|her|hers|herself)\b", re.IGNORECASE)
_PRONOUN_LABELS = {
    "he",
    "she",
    "they",
    "them",
    "theirs",
    "their",
    "themselves",
    "him",
    "his",
    "himself",
    "her",
    "hers",
    "herself",
    "we",
    "us",
    "our",
    "ours",
    "ourselves",
    "i",
    "me",
    "my",
    "mine",
    "myself",
    "you",
    "your",
    "yours",
    "yourself",
    "yourselves",
}

_CONFIDENCE_RANK = {"low": 1, "medium": 2, "high": 3}

_FEMALE_TITLE_HINTS = (
    "madame",
    "mme",
    "madam",
    "mrs",
    "miss",
    "ms",
    "lady",
    "countess",
    "baroness",
    "princess",
    "queen",
    "mademoiselle",
)

_MALE_TITLE_HINTS = (
    "monsieur",
    "m.",
    "mr",
    "sir",
    "lord",
    "count",
    "baron",
    "prince",
    "king",
    "abbé",
    "abbe",
)

_MALE_TOKEN_WEIGHTS = {
    "he": 1.0,
    "him": 0.6,
    "his": 0.75,
    "himself": 1.0,
}

_FEMALE_TOKEN_WEIGHTS = {
    "she": 1.0,
    "her": 0.4,
    "hers": 0.75,
    "herself": 1.0,
}

_STOP_LABELS = {
    "and",
    "but",
    "then",
    "though",
    "meanwhile",
    "therefore",
    "after",
    "before",
    "when",
    "while",
    "because",
    "as",
    "yet",
    "nor",
    "so",
    "thus",
    "suddenly",
    "eventually",
    "finally",
    "until",
    "unless",
}


@dataclass(slots=True)
class SpeakerGuess:
    speaker_id: str
    label: str
    count: int = 0
    confidence: str = "low"
    sample_quotes: List[Dict[str, str]] = field(default_factory=list)
    suppressed: bool = False
    gender: str = "unknown"
    detected_gender: str = "unknown"
    male_votes: int = 0
    female_votes: int = 0

    def register_occurrence(
        self,
        confidence: str,
        text: str,
        quote: Optional[str],
        male_votes: int,
        female_votes: int,
        sample_excerpt: Optional[str] = None,
    ) -> None:
        self.count += 1
        if _CONFIDENCE_RANK.get(confidence, 0) > _CONFIDENCE_RANK.get(
            self.confidence, 0
        ):
            self.confidence = confidence

        excerpt = (
            sample_excerpt
            if sample_excerpt is not None
            else _build_excerpt(text, quote)
        )
        gender_hint = _format_gender_hint(male_votes, female_votes)
        if excerpt:
            payload = {"excerpt": excerpt, "gender_hint": gender_hint}
            if payload not in self.sample_quotes:
                self.sample_quotes.append(payload)
                if len(self.sample_quotes) > 3:
                    self.sample_quotes = self.sample_quotes[:3]

        if male_votes:
            self.male_votes += male_votes
        if female_votes:
            self.female_votes += female_votes
        self.detected_gender = _derive_gender(
            self.male_votes, self.female_votes, self.detected_gender
        )
        if self.gender in {"unknown", "male", "female"}:
            self.gender = _derive_gender(
                self.male_votes, self.female_votes, self.gender
            )

    def as_dict(self) -> Dict[str, Any]:
        return {
            "id": self.speaker_id,
            "label": self.label,
            "count": self.count,
            "confidence": self.confidence,
            "sample_quotes": [dict(sample) for sample in self.sample_quotes],
            "suppressed": self.suppressed,
            "gender": self.gender,
            "detected_gender": self.detected_gender,
        }


@dataclass(slots=True)
class SpeakerAnalysis:
    assignments: Dict[str, str]
    speakers: Dict[str, SpeakerGuess]
    suppressed: List[str]
    narrator: str = "narrator"
    version: str = "1.0"
    stats: Dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> Dict[str, Any]:
        return {
            "version": self.version,
            "narrator": self.narrator,
            "assignments": dict(self.assignments),
            "speakers": {
                speaker_id: guess.as_dict()
                for speaker_id, guess in self.speakers.items()
            },
            "suppressed": list(self.suppressed),
            "stats": dict(self.stats),
        }


def analyze_speakers(
    chapters: Sequence[Dict[str, Any]] | Iterable[Dict[str, Any]],
    chunks: Sequence[Dict[str, Any]] | Iterable[Dict[str, Any]],
    *,
    threshold: int = 3,
    max_speakers: int = 8,
) -> SpeakerAnalysis:
    narrator_id = "narrator"
    speaker_guesses: Dict[str, SpeakerGuess] = {
        narrator_id: SpeakerGuess(
            speaker_id=narrator_id, label="Narrator", confidence="low"
        )
    }
    label_index: Dict[str, str] = {"Narrator": narrator_id}
    assignments: Dict[str, str] = {}
    suppressed: List[str] = []

    ordered_chunks = sorted(
        (dict(chunk) for chunk in chunks),
        key=lambda entry: (
            _safe_int(entry.get("chapter_index")),
            _safe_int(entry.get("chunk_index")),
        ),
    )
    last_explicit: Optional[str] = None
    explicit_assignments = 0
    unique_speakers: set[str] = set()

    for index, chunk in enumerate(ordered_chunks):
        chunk_id = str(chunk.get("id") or "")
        text = _get_chunk_text(chunk)
        speaker_id, confidence, quote = _infer_chunk_speaker(text, last_explicit)
        if speaker_id is None:
            speaker_id = last_explicit or narrator_id
            confidence = "medium" if last_explicit else "low"
            quote = quote or _extract_quote(text)
        if speaker_id != narrator_id:
            last_explicit = speaker_id
            explicit_assignments += 1

        if speaker_id in speaker_guesses:
            record_id = speaker_id
            guess = speaker_guesses[record_id]
            label = guess.label
        else:
            label = _normalize_label(speaker_id)
            record_id = label_index.get(label)
            if record_id is None:
                record_id = _dedupe_slug(_slugify(label), speaker_guesses)
                label_index[label] = record_id
                speaker_guesses[record_id] = SpeakerGuess(
                    speaker_id=record_id, label=label
                )
            guess = speaker_guesses[record_id]
        assignments[chunk_id] = record_id
        unique_speakers.add(record_id)

        if (
            record_id != narrator_id
            and record_id != speaker_id
            and speaker_id == last_explicit
        ):
            last_explicit = record_id

        sample_excerpt = None
        if record_id != narrator_id:
            sample_excerpt = _select_sample_excerpt(
                ordered_chunks, index, guess.label, quote, confidence
            )

        male_votes, female_votes = _count_gender_votes(text, guess.label)

        guess.register_occurrence(
            confidence, text, quote, male_votes, female_votes, sample_excerpt
        )

    active_speakers = [sid for sid in speaker_guesses if sid != narrator_id]
    # Apply minimum occurrence threshold.
    for speaker_id in list(active_speakers):
        guess = speaker_guesses[speaker_id]
        if guess.count < max(1, threshold):
            guess.suppressed = True
            suppressed.append(speaker_id)
            _reassign(assignments, speaker_id, narrator_id)
            active_speakers.remove(speaker_id)

    # Apply maximum active speaker cap.
    if max_speakers and len(active_speakers) > max_speakers:
        active_speakers.sort(key=lambda sid: (-speaker_guesses[sid].count, sid))
        for speaker_id in active_speakers[max_speakers:]:
            guess = speaker_guesses[speaker_id]
            guess.suppressed = True
            suppressed.append(speaker_id)
            _reassign(assignments, speaker_id, narrator_id)
        active_speakers = active_speakers[:max_speakers]

    narrator_guess = speaker_guesses[narrator_id]
    narrator_guess.count = sum(
        1 for value in assignments.values() if value == narrator_id
    )
    narrator_guess.confidence = "low"

    stats = {
        "total_chunks": len(ordered_chunks),
        "explicit_chunks": explicit_assignments,
        "active_speakers": len(active_speakers),
        "unique_speakers": len(unique_speakers),
        "suppressed": len(suppressed),
    }

    return SpeakerAnalysis(
        assignments=assignments,
        speakers=speaker_guesses,
        suppressed=suppressed,
        narrator=narrator_id,
        stats=stats,
    )


def _infer_chunk_speaker(
    text: str, last_explicit: Optional[str]
) -> Tuple[Optional[str], str, Optional[str]]:
    normalized = text.strip()
    if not normalized:
        return None, "low", None

    colon_match = _COLON_PATTERN.match(normalized)
    if colon_match:
        raw_label = colon_match.group(1)
        cleaned = _normalize_candidate_name(raw_label)
        if cleaned is None:
            return None, "low", colon_match.group(2).strip()
        quote = colon_match.group(2).strip()
        return cleaned, "high", quote

    quote = _extract_quote(normalized)
    if not quote:
        return None, "low", None

    before, after = _split_around_quote(normalized, quote)

    candidate = _match_name_near_quote(before, after)
    if candidate:
        cleaned = _normalize_candidate_name(candidate)
        if cleaned:
            return cleaned, "high", quote

    if last_explicit:
        pronoun_after = _PRONOUN_PATTERN.search(after)
        pronoun_before = _PRONOUN_PATTERN.search(before)
        if pronoun_after or pronoun_before:
            return last_explicit, "medium", quote

    return None, "low", quote


def _split_around_quote(text: str, quote: str) -> Tuple[str, str]:
    quote_index = text.find(quote)
    if quote_index == -1:
        return text, ""
    before = text[:quote_index]
    after = text[quote_index + len(quote) :]
    return before, after


def _match_name_near_quote(before: str, after: str) -> Optional[str]:
    trailing = before[-120:]
    leading = after[:120]

    match = _NAME_BEFORE_VERB.search(trailing)
    if match:
        name = match.group(1)
        if _looks_like_name(name):
            return name

    match = re.search(
        rf"({_NAME_PATTERN})\s*,?\s*{_VERB_PATTERN}", leading, flags=re.IGNORECASE
    )
    if match:
        name = match.group(1)
        if _looks_like_name(name):
            return name

    match = _VERB_BEFORE_NAME.search(leading)
    if match:
        name = match.group(1)
        if _looks_like_name(name):
            return name

    return None


def _looks_like_name(value: str) -> bool:
    normalized = _normalize_candidate_name(value)
    if not normalized:
        return False
    parts = normalized.split()
    if not parts:
        return False
    return all(part and part[0].isupper() for part in parts)


def _extract_quote(text: str) -> Optional[str]:
    match = _QUOTE_PATTERN.search(text)
    if not match:
        return None
    return match.group(0)


def _slugify(label: str) -> str:
    slug = re.sub(r"[^a-z0-9]+", "_", label.lower()).strip("_")
    return slug or "speaker"


def _dedupe_slug(slug: str, existing: Dict[str, SpeakerGuess]) -> str:
    candidate = slug
    index = 2
    while candidate in existing:
        candidate = f"{slug}_{index}"
        index += 1
    return candidate


def _normalize_label(label: str) -> str:
    words = re.split(r"\s+", label.strip())
    return " ".join(word.capitalize() for word in words if word)


def _safe_int(value: Any, default: int = 0) -> int:
    try:
        return int(value)
    except (TypeError, ValueError):
        return default


def _reassign(assignments: Dict[str, str], old: str, new: str) -> None:
    for key, value in list(assignments.items()):
        if value == old:
            assignments[key] = new


def _strip_diacritics(value: str) -> str:
    normalized = unicodedata.normalize("NFKD", value)
    return "".join(char for char in normalized if not unicodedata.combining(char))


def _count_gender_votes(text: str, label: Optional[str]) -> Tuple[int, int]:
    if not text:
        return 0, 0

    search_text = text
    windows: List[Tuple[int, int]] = []
    degrade_factor = 1.0

    if label:
        pattern = re.compile(re.escape(label), re.IGNORECASE)
        matches = list(pattern.finditer(search_text))
        if not matches:
            alt_label = _strip_diacritics(label)
            if alt_label and alt_label != label:
                ascii_text = _strip_diacritics(search_text)
                pattern_alt = re.compile(re.escape(alt_label), re.IGNORECASE)
                windows = [match.span() for match in pattern_alt.finditer(ascii_text)]
                # Map spans back roughly using proportional index
                if windows:
                    mapped: List[Tuple[int, int]] = []
                    for start, end in windows:
                        start_idx = min(
                            len(search_text) - 1,
                            int(start * len(search_text) / max(len(ascii_text), 1)),
                        )
                        end_idx = min(
                            len(search_text),
                            int(end * len(search_text) / max(len(ascii_text), 1)),
                        )
                        mapped.append((start_idx, end_idx))
                    windows = mapped
        else:
            windows = [match.span() for match in matches]

    if not windows:
        windows = [(0, len(search_text))]
        degrade_factor = 0.25

    radius = 60
    quote_spans: List[Tuple[int, int, str]] = []
    for match in _QUOTE_PATTERN.finditer(search_text):
        try:
            content_start, content_end = match.span(1)
        except IndexError:
            content_start, content_end = match.span()
        if content_start < content_end:
            quote_spans.append(
                (content_start, content_end, search_text[content_start:content_end])
            )

    normalized_label = _normalize_candidate_name(label) if label else None
    normalized_label_lower = normalized_label.lower() if normalized_label else None

    def _window_weight(position: int) -> float:
        for start, end in windows:
            if position < start - radius or position > end + radius:
                continue
            if position >= end:
                return 1.0
            if position <= start:
                return 0.2
            return 1.0
        return 0.0

    def _quote_weight(position: int) -> float:
        for start, end, content in quote_spans:
            if position < start or position >= end:
                continue
            local_index = position - start
            prefix = content[:local_index]
            tail = prefix[-80:]
            name_matches = list(re.finditer(_NAME_PATTERN, tail))
            if name_matches:
                last_name = _normalize_candidate_name(name_matches[-1].group(0))
                if (
                    normalized_label_lower
                    and last_name
                    and last_name.lower() == normalized_label_lower
                ):
                    return 0.6
                return 0.05
            if re.search(r"[.!?]\s", prefix):
                return 0.2
            if prefix.strip():
                return 0.15
            return 0.1
        return 1.0

    male_score = 0.0
    for match in _MALE_PRONOUN_PATTERN.finditer(search_text):
        base_weight = _window_weight(match.start())
        if not base_weight:
            continue
        quote_modifier = _quote_weight(match.start())
        weight = base_weight * quote_modifier
        if not weight:
            continue
        token = match.group(0).lower()
        male_score += _MALE_TOKEN_WEIGHTS.get(token, 0.6) * weight

    female_score = 0.0
    for match in _FEMALE_PRONOUN_PATTERN.finditer(search_text):
        base_weight = _window_weight(match.start())
        if not base_weight:
            continue
        quote_modifier = _quote_weight(match.start())
        weight = base_weight * quote_modifier
        if not weight:
            continue
        if quote_modifier >= 0.95:
            weight = max(weight, 0.4)
        token = match.group(0).lower()
        female_score += _FEMALE_TOKEN_WEIGHTS.get(token, 0.4) * weight

    for start, end in windows:
        span_start = max(0, start - 40)
        span_end = min(len(search_text), end + 40)
        span_text = search_text[span_start:span_end].lower()
        if any(title in span_text for title in _FEMALE_TITLE_HINTS):
            female_score += 2.5
        if any(title in span_text for title in _MALE_TITLE_HINTS):
            male_score += 2.5

    male_votes = int(round(male_score * degrade_factor))
    female_votes = int(round(female_score * degrade_factor))
    return male_votes, female_votes


def _derive_gender(male_votes: int, female_votes: int, current: str) -> str:
    if male_votes == 0 and female_votes == 0:
        return current if current != "unknown" else "unknown"

    male_threshold = max(2, female_votes + 1)
    female_threshold = max(2, male_votes + 1)

    if male_votes >= male_threshold:
        return "male"
    if female_votes >= female_threshold:
        return "female"

    if current in {"male", "female"}:
        return current
    return "unknown"


def _get_chunk_text(chunk: Dict[str, Any]) -> str:
    if not isinstance(chunk, dict):
        return ""
    value = chunk.get("normalized_text") or chunk.get("text") or ""
    return str(value)


def _trim_paragraph(paragraph: str, limit: int = 600) -> str:
    normalized = (paragraph or "").strip()
    if not normalized:
        return ""
    if len(normalized) <= limit:
        return normalized
    return normalized[: limit - 1].rstrip() + "…"


def _compose_context_excerpt(before: str, current: str, after: str) -> str:
    segments = []
    for value in (before, current, after):
        trimmed = _trim_paragraph(value)
        if trimmed:
            segments.append(trimmed)
    return "\n\n".join(segments)


def _contains_dialogue_attribution(label: str, text: str, quote: Optional[str]) -> bool:
    if not label or not text:
        return False
    escaped_label = re.escape(label)
    direct_pattern = re.compile(
        rf"\b{escaped_label}\b\s+(?:{_VERB_PATTERN})\b", re.IGNORECASE
    )
    reverse_pattern = re.compile(
        rf"(?:{_VERB_PATTERN})\s+\b{escaped_label}\b", re.IGNORECASE
    )
    colon_pattern = re.compile(rf"^\s*{escaped_label}\s*:\s*", re.IGNORECASE)

    if colon_pattern.search(text):
        return True
    if direct_pattern.search(text) or reverse_pattern.search(text):
        return True
    if quote:
        before, after = _split_around_quote(text, quote)
        if direct_pattern.search(before) or reverse_pattern.search(after):
            return True
    return False


def _select_sample_excerpt(
    chunks: Sequence[Dict[str, Any]],
    index: int,
    label: str,
    quote: Optional[str],
    confidence: str,
) -> Optional[str]:
    if confidence != "high" or not label:
        return None
    if index < 0 or index >= len(chunks):
        return None
    current = _get_chunk_text(chunks[index])
    if not current or not _contains_dialogue_attribution(label, current, quote):
        return None
    previous = _get_chunk_text(chunks[index - 1]) if index > 0 else ""
    following = _get_chunk_text(chunks[index + 1]) if index + 1 < len(chunks) else ""
    excerpt = _compose_context_excerpt(previous, current, following)
    return excerpt or None


def _build_excerpt(text: str, quote: Optional[str]) -> str:
    normalized = (text or "").strip()
    if not normalized:
        return ""
    if quote:
        location = normalized.find(quote)
        if location != -1:
            start = max(0, location - 120)
            end = min(len(normalized), location + len(quote) + 120)
            snippet = normalized[start:end].strip()
            if start > 0:
                snippet = "…" + snippet
            if end < len(normalized):
                snippet = snippet + "…"
            return snippet
    if len(normalized) > 240:
        return normalized[:240].rstrip() + "…"
    return normalized


def _format_gender_hint(male_votes: int, female_votes: int) -> str:
    if male_votes and female_votes:
        return "Context mentions both male and female pronouns."
    if male_votes:
        if male_votes >= 3:
            return "Multiple male pronouns detected nearby."
        return "Some male pronouns detected in the surrounding text."
    if female_votes:
        if female_votes >= 3:
            return "Multiple female pronouns detected nearby."
        return "Some female pronouns detected in the surrounding text."
    return "No clear pronoun signal detected."


def _normalize_candidate_name(raw: str) -> Optional[str]:
    if not raw:
        return None
    cleaned = raw.strip().strip("\"“”'’.,:;!")
    cleaned = re.sub(r"\s+", " ", cleaned).strip()
    if not cleaned:
        return None
    parts = cleaned.split()
    filtered: List[str] = []
    for part in parts:
        if not part:
            continue
        if not filtered and part.lower() in _STOP_LABELS:
            continue
        filtered.append(part)
    while filtered and filtered[-1].lower() in _STOP_LABELS:
        filtered.pop()
    if not filtered:
        return None
    if all(part.lower() in _STOP_LABELS for part in filtered):
        return None
    contiguous: List[str] = []
    for part in filtered:
        if part and part[0].isupper():
            contiguous.append(part)
        else:
            break
    if contiguous:
        candidate = " ".join(contiguous)
    else:
        candidate = ""
    if not candidate:
        return None
    lowered = candidate.lower()
    if lowered in _PRONOUN_LABELS or lowered in _STOP_LABELS:
        return None
    return candidate