Spaces:
Sleeping
Sleeping
| """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 | |