ingestion/transcript.py

import logging

import httpx

from ingestion.alphavantage import AV_BASE, _next_key, _key_count, _is_rate_limited

logger = logging.getLogger(__name__)


def _parse_transcript(data: dict) -> str:
    raw = data.get("transcript", "")
    if isinstance(raw, list):
        lines = []
        for item in raw:
            speaker = item.get("speaker", "")
            title = item.get("title", "")
            label = f"{speaker} ({title})" if title else speaker
            lines.append(f"{label}: {item.get('content', '')}")
        return "\n".join(lines)
    if raw is None:
        return ""
    return str(raw)


def fetch_transcript(ticker: str, quarter: str) -> str:
    """Fetch earnings call transcript from Alpha Vantage.

    Args:
        ticker: Stock symbol e.g. "AAPL"
        quarter: Alpha Vantage format e.g. "2024Q4"

    Returns:
        Transcript text, or empty string if unavailable.
    """
    rate_limited_keys = 0
    n_keys = max(_key_count(), 1)

    for attempt in range(n_keys):
        key = _next_key()
        key_label = f"key[...{key[-4:]}]"
        try:
            resp = httpx.get(
                AV_BASE,
                params={
                    "function": "EARNINGS_CALL_TRANSCRIPT",
                    "symbol": ticker.upper(),
                    "quarter": quarter,
                    "apikey": key,
                },
                timeout=30,
            )
            resp.raise_for_status()
            data = resp.json()

            if "Information" in data:
                rate_limited_keys += 1
                logger.warning(
                    "[transcript] %s %s %s — rate limit hit (attempt %d/%d): %s",
                    ticker, quarter, key_label, attempt + 1, n_keys, data["Information"],
                )
                continue

            if "Note" in data:
                rate_limited_keys += 1
                logger.warning(
                    "[transcript] %s %s %s — API note/rate limit (attempt %d/%d): %s",
                    ticker, quarter, key_label, attempt + 1, n_keys, data["Note"],
                )
                continue

            if "Error Message" in data:
                logger.warning(
                    "[transcript] %s %s %s — API error: %s",
                    ticker, quarter, key_label, data["Error Message"],
                )
                return ""

            text = _parse_transcript(data)
            if text:
                logger.info(
                    "[transcript] %s %s — fetched %d chars via %s",
                    ticker, quarter, len(text), key_label,
                )
            else:
                logger.info(
                    "[transcript] %s %s — API returned empty transcript (no content available) via %s",
                    ticker, quarter, key_label,
                )
            return text

        except ValueError:
            raise
        except Exception as e:
            logger.warning(
                "[transcript] %s %s %s — request error (attempt %d/%d): %s",
                ticker, quarter, key_label, attempt + 1, n_keys, e,
            )
            continue

    if rate_limited_keys == n_keys:
        logger.error(
            "[transcript] %s %s — ALL %d keys exhausted by rate limits. No transcript fetched.",
            ticker, quarter, n_keys,
        )
    else:
        logger.error(
            "[transcript] %s %s — failed after %d attempt(s). No transcript fetched.",
            ticker, quarter, n_keys,
        )
    return ""