storage/sections_db.py

"""storage/sections_db.py — persist raw section text across filing periods.

Stores verbatim Business, Segments/Geography, MD&A, Risk Factors, and
transcript text keyed by (ticker, period, section). Used by company-profile
generation and analysis/textdiff.py without re-fetching source documents.

This is append-only at ingest time; textdiff reads it at runtime.
"""
from __future__ import annotations

import sqlite3
from pathlib import Path
from typing import Optional

SECTIONS_DB_PATH = Path("data/sections.db")

_SCHEMA = """
CREATE TABLE IF NOT EXISTS sections (
    ticker      TEXT NOT NULL,
    period      TEXT NOT NULL,
    form_type   TEXT NOT NULL,
    section     TEXT NOT NULL,
    text        TEXT NOT NULL DEFAULT '',
    ingested_at TEXT,
    PRIMARY KEY (ticker, period, section)
)
"""


def init_sections_db() -> None:
    SECTIONS_DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    with sqlite3.connect(SECTIONS_DB_PATH) as conn:
        conn.execute(_SCHEMA)


def upsert_section(
    ticker: str,
    period: str,
    form_type: str,
    section: str,
    text: str,
) -> None:
    """Write or overwrite a supported filing or transcript section."""
    from datetime import datetime, timezone
    SECTIONS_DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    with sqlite3.connect(SECTIONS_DB_PATH) as conn:
        conn.execute(_SCHEMA)
        conn.execute(
            """
            INSERT INTO sections (ticker, period, form_type, section, text, ingested_at)
            VALUES (?, ?, ?, ?, ?, ?)
            ON CONFLICT(ticker, period, section) DO UPDATE SET
                form_type   = excluded.form_type,
                text        = excluded.text,
                ingested_at = excluded.ingested_at
            """,
            (
                ticker.upper(),
                period,
                form_type,
                section,
                text or "",
                datetime.now(timezone.utc).isoformat(),
            ),
        )


def get_section(ticker: str, period: str, section: str) -> Optional[str]:
    """Return the stored text for (ticker, period, section), or None if absent."""
    if not SECTIONS_DB_PATH.exists():
        return None
    with sqlite3.connect(SECTIONS_DB_PATH) as conn:
        row = conn.execute(
            "SELECT text FROM sections WHERE ticker=? AND period=? AND section=?",
            (ticker.upper(), period, section),
        ).fetchone()
    return row[0] if row else None


def _period_sort_key(period: str) -> tuple[int, int]:
    """Parse 'Q12027' → (2027, 1) for correct chronological sort (newest first).

    Falls back to (0, 0) for unparseable strings (e.g. 'FY2024').
    """
    if not period:
        return (0, 0)
    if period.startswith("Q") and len(period) >= 6:
        try:
            body = period[1:]          # "12027"
            year = int(body[-4:])      # 2027
            quarter = int(body[:-4])   # 1
            return (year, quarter)
        except (ValueError, IndexError):
            pass
    if period.startswith("FY") and len(period) == 6:
        try:
            return (int(period[2:]), 0)
        except ValueError:
            pass
    return (0, 0)


def _transcript_sort_key(period: str) -> tuple[int, int]:
    """Chronological sort key that places FY calls as Q4 of their year.

    'Q12026' → (2026, 1); 'FY2025' → (2025, 4) so the Q4/FY earnings call
    sorts between Q32025 and Q12026. Unparseable strings → (0, 0).
    """
    if period and period.startswith("FY") and len(period) == 6:
        try:
            return (int(period[2:]), 4)
        except ValueError:
            return (0, 0)
    return _period_sort_key(period)


def get_recent_transcripts(ticker: str, n: int = 4) -> list[tuple[str, str]]:
    """Return [(period, text)] for the n most recent non-empty transcripts.

    Mixes 10-Q and 10-K (FY) earnings calls; FY periods sort as Q4 of their
    year. Results are in chronological order (oldest first).
    """
    if not SECTIONS_DB_PATH.exists():
        return []
    with sqlite3.connect(SECTIONS_DB_PATH) as conn:
        rows = conn.execute(
            "SELECT period, text FROM sections "
            "WHERE ticker=? AND section='transcript' AND length(text) > 0",
            (ticker.upper(),),
        ).fetchall()
    rows.sort(key=lambda r: _transcript_sort_key(r[0]))
    return [(r[0], r[1]) for r in rows[-n:]]


def get_periods_for_ticker(ticker: str, form_type: Optional[str] = None) -> list[str]:
    """Return all period strings stored for a ticker, sorted newest first (chronologically).

    Optionally filtered by form_type (e.g. '10-Q').
    """
    if not SECTIONS_DB_PATH.exists():
        return []
    with sqlite3.connect(SECTIONS_DB_PATH) as conn:
        if form_type:
            rows = conn.execute(
                "SELECT DISTINCT period FROM sections WHERE ticker=? AND form_type=?",
                (ticker.upper(), form_type),
            ).fetchall()
        else:
            rows = conn.execute(
                "SELECT DISTINCT period FROM sections WHERE ticker=?",
                (ticker.upper(),),
            ).fetchall()
    periods = [r[0] for r in rows]
    periods.sort(key=_period_sort_key, reverse=True)
    return periods