analysis/transcript_parse.py

"""analysis/transcript_parse.py — structural parsing of flattened earnings-call text.

Pure regex, zero embeddings, zero LLM. Operates on the Alpha Vantage transcript
format produced by ingestion/transcript.py: one line per speaker segment,
either "Speaker: content" or "Speaker (Title): content".

Reconstructs:
  - prepared remarks vs Q&A boundary
  - speaker roles (operator / management / analyst)
  - analyst question → management answer exchanges

Degrades gracefully: if the structure cannot be recovered, the full text is
treated as management speech and the Q&A exchange list is empty.
"""
from __future__ import annotations

import re
from dataclasses import dataclass, field

# "Speaker: content" or "Speaker (Title): content" at line start.
_SPEAKER_RE = re.compile(r"^([A-Z][\w.\-' ]{0,60}?)(?:\s*\(([^)]{1,60})\))?:\s+(.*)$")

# Marks the transition from prepared remarks to analyst Q&A.
_QA_BOUNDARY_RE = re.compile(
    r"question-and-answer session"
    r"|question and answer session"
    r"|ready to (?:start|begin) the q\s*&\s*a"
    r"|open (?:up )?the (?:call|line|floor)s? for questions"
    r"|poll (?:the audience|the lines?|for) ?(?:for questions|questions)?"
    r"|(?:take|taking) (?:your|the first) questions?"
    r"|first question comes? from",
    re.IGNORECASE,
)

# Operator hand-offs that name the analyst and their firm.
_ANALYST_INTRO_RE = re.compile(
    r"(?:line of|comes? from(?: the line of)?)\s+([A-Z][\w.\-' ]+?)\s+(?:with|from|at)\s+",
)

_MIN_SEGMENTS = 5


@dataclass
class Segment:
    speaker: str
    text: str
    role: str = "unknown"  # operator | management | analyst | unknown


@dataclass
class QAExchange:
    analyst: str
    question: str
    answer: str


@dataclass
class ParsedCall:
    period: str
    prepared_text: str       # management prepared remarks (pre-Q&A)
    management_text: str     # prepared remarks + all answers
    qa: list[QAExchange] = field(default_factory=list)
    n_segments: int = 0


def _last_name(name: str) -> str:
    parts = name.strip().rstrip(".").split()
    return parts[-1].lower() if parts else ""


def _split_segments(text: str) -> list[Segment]:
    segments: list[Segment] = []
    for line in text.splitlines():
        line = line.strip()
        if not line:
            continue
        m = _SPEAKER_RE.match(line)
        if m:
            title = m.group(2) or ""
            speaker = m.group(1).strip()
            seg = Segment(speaker=speaker, text=m.group(3).strip())
            if title and re.search(r"analyst", title, re.IGNORECASE):
                seg.role = "analyst"
            segments.append(seg)
        elif segments:
            segments[-1].text += " " + line
    return segments


def parse_call(period: str, text: str) -> ParsedCall:
    """Parse one flattened transcript into roles and Q&A exchanges.

    Never raises. Falls back to a Q&A-free ParsedCall whose prepared/management
    text is the full transcript when structure cannot be recovered.
    """
    segments = _split_segments(text or "")
    if len(segments) < _MIN_SEGMENTS:
        full = (text or "").strip()
        return ParsedCall(
            period=period, prepared_text=full, management_text=full,
            qa=[], n_segments=len(segments),
        )

    # Locate the prepared-remarks → Q&A boundary. Operator intros often
    # announce "there will be a question-and-answer session" before anyone
    # has spoken — only accept a boundary once a non-operator segment exists.
    qa_start: int | None = None
    for i, seg in enumerate(segments):
        if not _QA_BOUNDARY_RE.search(seg.text):
            continue
        has_speech_before = any(
            s.speaker.lower() != "operator" for s in segments[:i]
        )
        if has_speech_before:
            qa_start = i
            break

    # Analyst roster from Operator hand-offs (anywhere in the call).
    roster: set[str] = set()
    for seg in segments:
        if seg.speaker.lower() == "operator":
            seg.role = "operator"
            for m in _ANALYST_INTRO_RE.finditer(seg.text):
                roster.add(_last_name(m.group(1)))

    # Management = non-operator speakers heard before the Q&A boundary.
    pre_qa_end = qa_start if qa_start is not None else len(segments)
    management: set[str] = {
        _last_name(seg.speaker)
        for seg in segments[:pre_qa_end]
        if seg.role not in ("operator", "analyst") and seg.speaker
    }

    # Assign roles.
    prev_role = ""
    for i, seg in enumerate(segments):
        if seg.role in ("operator", "analyst"):
            prev_role = seg.role
            continue
        key = _last_name(seg.speaker)
        if key in management:
            seg.role = "management"
        elif key in roster:
            seg.role = "analyst"
        elif qa_start is not None and i > qa_start:
            # Unknown speaker in Q&A: question-shaped or operator hand-off → analyst.
            if seg.text.rstrip().endswith("?") or prev_role == "operator":
                seg.role = "analyst"
            else:
                seg.role = "management"
        else:
            seg.role = "management"
        prev_role = seg.role

    prepared_parts = [
        seg.text for seg in segments[:pre_qa_end] if seg.role == "management"
    ]
    management_parts = [seg.text for seg in segments if seg.role == "management"]

    # Pair analyst turns with the management turns that follow them.
    qa: list[QAExchange] = []
    if qa_start is not None:
        current: QAExchange | None = None
        for seg in segments[qa_start:]:
            if seg.role == "analyst":
                if current is None or current.answer:
                    if current is not None and current.answer:
                        qa.append(current)
                    current = QAExchange(analyst=seg.speaker, question=seg.text, answer="")
                else:
                    current.question += " " + seg.text
            elif seg.role == "management" and current is not None:
                current.answer = (current.answer + " " + seg.text).strip()
            elif seg.role == "operator" and current is not None and current.answer:
                qa.append(current)
                current = None
        if current is not None and current.answer:
            qa.append(current)

    return ParsedCall(
        period=period,
        prepared_text="\n".join(prepared_parts).strip(),
        management_text="\n".join(management_parts).strip(),
        qa=qa,
        n_segments=len(segments),
    )