backend/src/finrag/ingestion/parse.py

"""Parse downloaded 10-K HTML files into chunks ready for embedding.

Pipeline:
    data/raw/{TICKER}_{FY}/filing.htm + metadata.json
        ──▶  partition_html  ──▶  list[Element]
        ──▶  segment by Title, split tables off
        ──▶  chunk_by_title within each section (narrative)
        ──▶  emit table elements as their own chunks
    data/processed/{TICKER}_{FY}.jsonl    (one Chunk per line)
    data/processed/manifest.json
"""

from __future__ import annotations

import hashlib
import json
from pathlib import Path
from typing import Any

from pydantic import BaseModel
# NOTE: `unstructured` is imported lazily inside parse_filing() (see below), not
# at module top. It's a heavy, ingestion-only dependency (torch/transformers via
# unstructured[pdf]) kept OUT of the runtime image — but this module is imported
# transitively at runtime for its constants (PROCESSED_DIR, Chunk), so a top-level
# import would crash the server. Install it for ingestion with `uv sync --group ingestion`.

# ── Paths ─────────────────────────────────────────────────────────────────
# parse.py → ingestion/ → finrag/ → src/ → backend/ → ROOT
REPO_ROOT = Path(__file__).resolve().parents[4]
RAW_DIR = REPO_ROOT / "data" / "raw"
PROCESSED_DIR = REPO_ROOT / "data" / "processed"

# ── Chunking knobs (see Decision 5 discussion for reasoning) ──────────────
# max:   hard ceiling. Stays comfortably under Cohere embed-v3's ~512 token limit.
# new_after: soft target — lets natural breaks happen before reaching max.
# combine_under: absorbs tiny stub chunks (lone section titles, footers).
# overlap: only kicks in when a section is split mid-section due to length.
MAX_CHARACTERS = 1500
NEW_AFTER_N_CHARS = 1200
COMBINE_TEXT_UNDER_N_CHARS = 200
OVERLAP = 150


# ── Models ────────────────────────────────────────────────────────────────
class Chunk(BaseModel):
    """One retrievable unit: either a narrative passage or a single table.

    Every field after `chunk_type` is provenance — copied from the filing's
    metadata.json so the chunk is self-contained when it lands in Qdrant.
    The retriever can filter on any of these fields without joining back.
    """

    chunk_id: str          # deterministic SHA-256 hash, 16 hex chars
    text: str              # the actual content to embed
    chunk_type: str        # "narrative" | "table"
    section_title: str | None
    section_path: list[str]
    # Filing provenance
    ticker: str
    company_name: str
    fiscal_year: int
    period_of_report: str  # ISO date string
    accession_number: str
    sec_url: str
    # Position within document — currently a monotonic ordinal per filing.
    # On Day 2+ this becomes the anchor for citation-viewer highlighting.
    element_index: int


# ── Helpers ───────────────────────────────────────────────────────────────
def _hash_chunk(ticker: str, fiscal_year: int, position: int, text: str) -> str:
    """Stable ID. Including `text` means changing chunking params produces
    new IDs rather than silently overwriting old vectors with new content."""
    h = hashlib.sha256()
    h.update(f"{ticker}|{fiscal_year}|{position}|".encode())
    h.update(text.encode())
    return h.hexdigest()[:16]


def _load_filing_metadata(filing_dir: Path) -> dict[str, Any]:
    return json.loads((filing_dir / "metadata.json").read_text())


def _element_category(el: Any) -> str:
    # Unstructured elements expose `.category`; fall back to class name.
    return getattr(el, "category", type(el).__name__)


def _table_text(el: Any) -> str:
    """Prefer the HTML representation — preserves rows/columns for the
    embedder. Falls back to flattened text if HTML isn't available."""
    md = getattr(el, "metadata", None)
    if md is not None:
        html = getattr(md, "text_as_html", None)
        if html:
            return html
    return el.text


# ── Core ──────────────────────────────────────────────────────────────────
def parse_filing(filing_dir: Path) -> list[Chunk]:
    """Read a filing directory, return chunks ready for embedding."""
    # Lazy import: keeps `unstructured` out of the runtime import path (it's only
    # needed here, during offline ingestion). Clear hint if the group is missing.
    try:
        from unstructured.chunking.title import chunk_by_title
        from unstructured.partition.html import partition_html
    except ModuleNotFoundError as e:  # pragma: no cover
        raise ModuleNotFoundError(
            "Ingestion requires the 'unstructured' extra. Install it with "
            "`uv sync --group ingestion` (it's excluded from the runtime image)."
        ) from e

    meta = _load_filing_metadata(filing_dir)
    htm_path = filing_dir / "filing.htm"

    # Stage 1 — atomize the HTML into typed elements.
    # `partition_html` is slow on first run (downloads NLTK data); fast after.
    elements = partition_html(filename=str(htm_path))

    # Stage 2a — walk elements once, building two parallel structures:
    #   - sections: a list of section buckets, each holding narrative elements
    #   - tables: pulled out into their own stream with section context attached
    #
    # Why bucket by section ourselves rather than relying on chunk_by_title's
    # implicit handling? Because we need clean `section_title` attribution per
    # chunk, and Unstructured's CompositeElement doesn't always expose the
    # underlying Title element reliably across versions.
    sections: list[dict[str, Any]] = []
    tables: list[dict[str, Any]] = []
    current_section: dict[str, Any] | None = None

    for idx, el in enumerate(elements):
        category = _element_category(el)

        if category == "Title":
            # Start a new section bucket. The Title element itself goes in
            # so chunk_by_title sees it as the leading boundary.
            current_section = {
                "title": el.text,
                "elements": [el],
            }
            sections.append(current_section)
        elif category == "Table":
            tables.append(
                {
                    "idx": idx,
                    "element": el,
                    "section_title": current_section["title"] if current_section else None,
                }
            )
        else:
            # Anything else: NarrativeText, ListItem, Header, etc.
            if current_section is None:
                # Content before the first Title (cover page, etc.)
                current_section = {"title": None, "elements": []}
                sections.append(current_section)
            current_section["elements"].append(el)

    chunks: list[Chunk] = []
    position = 0  # monotonic counter, used as element_index for stable IDs

    # Stage 2b — chunk narrative *within* each section.
    # By calling chunk_by_title per section, chunks never cross section
    # boundaries — a guarantee we couldn't make with a single document-wide call.
    for section in sections:
        if not section["elements"]:
            continue

        composite_chunks = chunk_by_title(
            section["elements"],
            max_characters=MAX_CHARACTERS,
            new_after_n_chars=NEW_AFTER_N_CHARS,
            combine_text_under_n_chars=COMBINE_TEXT_UNDER_N_CHARS,
            overlap=OVERLAP,
        )

        for cc in composite_chunks:
            text = cc.text.strip()
            if not text:
                continue  # skip empty composite results
            chunks.append(
                Chunk(
                    chunk_id=_hash_chunk(
                        meta["ticker"], meta["fiscal_year"], position, text
                    ),
                    text=text,
                    chunk_type="narrative",
                    section_title=section["title"],
                    section_path=[section["title"]] if section["title"] else [],
                    ticker=meta["ticker"],
                    company_name=meta["company_name"],
                    fiscal_year=meta["fiscal_year"],
                    period_of_report=meta["period_of_report"],
                    accession_number=meta["accession_number"],
                    sec_url=meta["sec_url"],
                    element_index=position,
                )
            )
            position += 1

    # Stage 2c — emit tables as their own chunks.
    # The text is the table's HTML (when available), which keeps cell/column
    # structure visible to the embedder. Tables that are very large will get
    # truncated by Cohere's 512-token limit — accepted, because Day 2's
    # DuckDB extractor will handle these structurally anyway.
    for tbl in tables:
        text = _table_text(tbl["element"]).strip()
        if not text:
            continue
        chunks.append(
            Chunk(
                chunk_id=_hash_chunk(
                    meta["ticker"], meta["fiscal_year"], position, text
                ),
                text=text,
                chunk_type="table",
                section_title=tbl["section_title"],
                section_path=[tbl["section_title"]] if tbl["section_title"] else [],
                ticker=meta["ticker"],
                company_name=meta["company_name"],
                fiscal_year=meta["fiscal_year"],
                period_of_report=meta["period_of_report"],
                accession_number=meta["accession_number"],
                sec_url=meta["sec_url"],
                element_index=position,
            )
        )
        position += 1

    return chunks


# ── CLI ───────────────────────────────────────────────────────────────────
def _write_chunks(chunks: list[Chunk], out_file: Path) -> None:
    with out_file.open("w", encoding="utf-8") as f:
        for c in chunks:
            f.write(c.model_dump_json() + "\n")


def _read_chunks(out_file: Path) -> list[Chunk]:
    return [
        Chunk.model_validate_json(line)
        for line in out_file.read_text(encoding="utf-8").splitlines()
        if line.strip()
    ]


def main() -> None:
    PROCESSED_DIR.mkdir(parents=True, exist_ok=True)
    filing_dirs = sorted(
        d for d in RAW_DIR.iterdir() if d.is_dir() and (d / "metadata.json").exists()
    )
    print(f"Found {len(filing_dirs)} filings to parse.\n")

    manifest_entries: list[dict[str, Any]] = []

    for filing_dir in filing_dirs:
        meta = _load_filing_metadata(filing_dir)
        ticker = meta["ticker"]
        fy = meta["fiscal_year"]
        out_file = PROCESSED_DIR / f"{ticker}_{fy}.jsonl"

        if out_file.exists():
            chunks = _read_chunks(out_file)
            print(f"  ↳ skip   {ticker} FY{fy} ({len(chunks)} chunks on disk)")
        else:
            print(f"  ↳ parse  {ticker} FY{fy}…", end="", flush=True)
            chunks = parse_filing(filing_dir)
            _write_chunks(chunks, out_file)
            print(f" → {len(chunks)} chunks")

        manifest_entries.append(
            {
                "ticker": ticker,
                "fiscal_year": fy,
                "chunks_total": len(chunks),
                "chunks_narrative": sum(1 for c in chunks if c.chunk_type == "narrative"),
                "chunks_table": sum(1 for c in chunks if c.chunk_type == "table"),
                "path": out_file.relative_to(REPO_ROOT).as_posix(),
            }
        )

    manifest_path = PROCESSED_DIR / "manifest.json"
    manifest_path.write_text(json.dumps({"filings": manifest_entries}, indent=2))
    total = sum(e["chunks_total"] for e in manifest_entries)
    print(f"\nDone. {total} chunks across {len(manifest_entries)} filings.")
    print(f"Manifest: {manifest_path}")


if __name__ == "__main__":
    main()