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

# A speaker's role routinely carries its own parentheses — the common form is
# "Jane Roe (Analyst (Example Bank)):". A title pattern that stops at the first
# closing bracket never reaches the colon on those lines, so the turn was not
# recognised as a speaker change at all and got appended to whoever spoke last.
# One level of nesting is enough for every format observed.
_TITLE = r"(?:[^()]|\([^()]*\))"

# "Speaker: content" or "Speaker (Title): content" at line start.
_SPEAKER_RE = re.compile(
    rf"^([A-Z][\w.\-' ]{{0,60}}?)(?:\s*\(({_TITLE}{{1,80}})\))?:\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
    # Answers alone, so a caller can tell what management volunteered from
    # what analysts made it address. Empty when no Q&A boundary was found.
    answers_text: str = ""
    # Every voice on the call, in order of first appearance. Term extraction
    # needs them: a name announced at each hand-off is not a subject.
    speakers: list[str] = field(default_factory=list)


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()
            for seg in _split_inline_speakers(speaker, title, m.group(3).strip()):
                segments.append(seg)
        elif segments:
            segments[-1].text += " " + line
    return segments


# Some providers put a whole exchange on one line: the operator's hand-off and
# the analyst's question share it, as in
#
#   Operator: Our first question comes from Jane Roe with Example Bank.
#   Jane Roe (Analyst (Example Bank)): I have two, one for...
#
# arriving as a single line. Matching only at line start then buried every
# analyst turn inside the operator's segment, so the call parsed to zero
# questions despite being fully structured. Requiring a multi-word capitalised
# name followed by a parenthesised role keeps ordinary prose ("the ratio (as
# defined): ...") from being mistaken for a speaker change.
_INLINE_SPEAKER_RE = re.compile(
    rf"(?<=\s)((?:[A-Z][\w.\-']*\s){{1,3}}[A-Z][\w.\-']*)\s*\(({_TITLE}{{1,80}})\):\s+"
)


def _split_inline_speakers(speaker: str, title: str, body: str) -> list[Segment]:
    """Split one line into every speaker turn it actually contains."""

    def _make(name: str, role_title: str, content: str) -> Segment:
        seg = Segment(speaker=name.strip(), text=content.strip())
        if role_title and re.search(r"analyst", role_title, re.IGNORECASE):
            seg.role = "analyst"
        return seg

    matches = list(_INLINE_SPEAKER_RE.finditer(body))
    if not matches:
        return [_make(speaker, title, body)]

    out = [_make(speaker, title, body[: matches[0].start()])]
    for index, match in enumerate(matches):
        end = matches[index + 1].start() if index + 1 < len(matches) else len(body)
        out.append(_make(match.group(1), match.group(2), body[match.end():end]))
    # A hand-off line can leave the operator with nothing but the introduction;
    # keep it only when it carries text of its own.
    return [seg for seg in out if seg.text]


def _ordered_speakers(segments: list[Segment]) -> list[str]:
    """Distinct speaker names, in order of first appearance."""
    seen: list[str] = []
    for seg in segments:
        name = seg.speaker.strip()
        if name and name not in seen:
            seen.append(name)
    return seen


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),
            speakers=_ordered_speakers(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

    # Structural fallback when no announcement phrase matched.
    #
    # Transcript providers word the hand-off differently, and some drop the
    # announcement entirely — Apple's calls are a standing example, where the
    # host moves straight to the first analyst. The section is still plainly
    # there in the structure: an operator turn, then a voice that has not
    # spoken yet. That new voice is the first analyst, so the operator turn
    # before it is the boundary. Requiring prior speech keeps an operator's
    # opening housekeeping from being mistaken for the Q&A.
    if qa_start is None:
        heard: set[str] = set()
        for i, seg in enumerate(segments):
            if seg.speaker.lower() != "operator":
                heard.add(_last_name(seg.speaker))
                continue
            if not heard:
                continue
            following = next(
                (s for s in segments[i + 1:] if s.speaker.lower() != "operator"),
                None,
            )
            if following is not None and _last_name(following.speaker) not in heard:
                qa_start = i
                break

    # Last resort: a call with no operator at all.
    #
    # Some issuers run their own Q&A — an investor-relations host reads the
    # questions and there is never an "Operator" turn to key off. Tesla's calls
    # are the standing example. What still holds is the shape: prepared remarks
    # from a handful of known voices, then a voice that has not been heard yet
    # asking something. The question mark is what separates that from a second
    # executive joining the prepared remarks.
    if qa_start is None:
        heard = set()
        for i, seg in enumerate(segments):
            name = _last_name(seg.speaker)
            # Two prior voices ≈ host plus at least one executive, i.e. the
            # prepared section is genuinely under way.
            if len(heard) >= 2 and name not in heard and "?" in seg.text:
                qa_start = i
                break
            heard.add(name)

    # 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"]
    answer_parts = [
        seg.text for seg in segments[pre_qa_end:] 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),
        answers_text="\n".join(answer_parts).strip(),
        speakers=_ordered_speakers(segments),
    )