All-in-one-Financial-Analysis/data/sec_downloader.py

"""
SEC 10-K download + section extraction: download via sec-edgar-downloader, extract items, cache.
"""
import tempfile
from pathlib import Path

from config.constants import (
    ITEM1A_PATTERNS, ITEM3_PATTERNS, ITEM7_PATTERNS, ITEM8_PATTERNS, ITEM9A_PATTERNS,
)
from data.sec_parser import (
    find_item_section_generic, _find_section_start,
    _extract_item_from_full, clean_text_for_llm, smart_chunk,
)
from data.sec_fetcher import (
    get_edgar_downloader, find_downloaded_10k_path, find_all_10k_filing_dirs,
    get_main_10k_text, _load_10k_html_from_cache, _get_main_10k_html_file,
    _save_10k_html_to_cache, _load_10k_from_cache, _save_10k_to_cache,
)


def download_and_extract_all_items(ticker: str, email: str) -> dict:
    """Download latest 10-K, extract Item 1A, 3, 7, 9A; clean and return (and optionally cache).
    Also saves the raw HTML file to data/TICKER_latest_raw.html for the native viewer."""
    Downloader = get_edgar_downloader()
    with tempfile.TemporaryDirectory() as tmpdir:
        download_root = Path(tmpdir)
        dl = Downloader("FQDC-10K-Analyzer", email, str(download_root))
        dl.get("10-K", ticker.upper(), limit=1, download_details=True)
        filing_dir = find_downloaded_10k_path(download_root, ticker)
        if not filing_dir:
            raise FileNotFoundError(f"Could not find 10-K for ticker '{ticker}'.")
        full_text = get_main_10k_text(filing_dir)
        if not full_text:
            raise ValueError("Could not extract text from the 10-K.")
        # Save raw HTML to persistent cache while temp dir is still open
        if not _load_10k_html_from_cache(ticker):
            main_html_path = _get_main_10k_html_file(filing_dir)
            if main_html_path:
                try:
                    with open(main_html_path, "r", encoding="utf-8", errors="replace") as _f:
                        _save_10k_html_to_cache(ticker, _f.read())
                except Exception:
                    pass
    item1a = find_item_section_generic(full_text, ITEM1A_PATTERNS, 1, ["Risk", "Factors"], max_chars=80000)
    item3 = _extract_item_from_full(full_text, ITEM3_PATTERNS, 3, ["Legal", "Proceedings"], max_chars=40000)
    item9a = _extract_item_from_full(full_text, ITEM9A_PATTERNS, 9, ["Controls", "Procedures", "Internal"], max_chars=40000)
    start7 = _find_section_start(full_text, ITEM7_PATTERNS, 7)
    text_after_7 = full_text[start7:] if start7 >= 0 else full_text
    item7 = find_item_section_generic(text_after_7, ITEM7_PATTERNS, 7, ["Management's Discussion", "MD&A", "Analysis"], max_chars=100000)
    if not item7 and text_after_7:
        item7 = text_after_7[:120000]
    item8 = _extract_item_from_full(full_text, ITEM8_PATTERNS, 8, ["Financial Statements", "Supplementary Data"], max_chars=200000)
    data = {
        "item1a": clean_text_for_llm(item1a or ""),
        "item3": clean_text_for_llm(item3 or ""),
        "item9a": clean_text_for_llm(item9a or ""),
        "item7": clean_text_for_llm(item7 or ""),
        "item8": clean_text_for_llm(item8 or ""),
    }
    _save_10k_to_cache(ticker, data)
    return data


def get_10k_sections(ticker: str, email: str) -> tuple:
    """Return (sections dict, status). status = 'cache' if loaded from file else 'downloaded'."""
    cached = _load_10k_from_cache(ticker)
    if cached is not None:
        return cached, "cache"
    return download_and_extract_all_items(ticker, email), "downloaded"


def download_and_extract_item7_and_1a(ticker: str, email: str) -> tuple:
    """Fetch 10-K and return full_text, Item 1A (Risk Factors), Item 7 (MD&A). Uses cache when available."""
    sections, _ = get_10k_sections(ticker, email)
    return "", sections.get("item1a", "") or "", sections.get("item7", "") or ""


def download_item7_latest_and_3y_ago(ticker: str, email: str) -> tuple:
    """Download up to 5 10-Ks; extract Item 1A (latest only) and Item 7 from latest and from 3 years ago."""
    Downloader = get_edgar_downloader()
    with tempfile.TemporaryDirectory() as tmpdir:
        download_root = Path(tmpdir)
        dl = Downloader("FQDC-10K-Analyzer", email, str(download_root))
        dl.get("10-K", ticker.upper(), limit=5, download_details=True)
        filing_dirs = find_all_10k_filing_dirs(download_root, ticker)
        if not filing_dirs:
            raise FileNotFoundError(f"Could not find 10-K for ticker '{ticker}'.")
        full_latest = get_main_10k_text(filing_dirs[0])
        if not full_latest:
            raise ValueError("Could not extract text from the latest 10-K.")
        item1a = find_item_section_generic(
            full_latest, ITEM1A_PATTERNS, 1, ["Risk", "Factors"], max_chars=80000
        )
        text_after_7 = full_latest[_find_section_start(full_latest, ITEM7_PATTERNS, 7):] if _find_section_start(full_latest, ITEM7_PATTERNS, 7) >= 0 else full_latest
        item7_latest = find_item_section_generic(
            text_after_7, ITEM7_PATTERNS, 7, ["Management's Discussion", "MD&A", "Analysis"], max_chars=100000
        )
        if not item7_latest and text_after_7:
            item7_latest = smart_chunk(text_after_7[:120000], max_chars=20000)
        item7_3y_ago = None
        has_comparison = False
        if len(filing_dirs) >= 4:
            full_3y = get_main_10k_text(filing_dirs[3])
            if full_3y:
                text_3y = full_3y[_find_section_start(full_3y, ITEM7_PATTERNS, 7):] if _find_section_start(full_3y, ITEM7_PATTERNS, 7) >= 0 else full_3y
                item7_3y_ago = find_item_section_generic(
                    text_3y, ITEM7_PATTERNS, 7, ["Management's Discussion", "MD&A", "Analysis"], max_chars=100000
                )
                if not item7_3y_ago and text_3y:
                    item7_3y_ago = smart_chunk(text_3y[:120000], max_chars=20000)
                has_comparison = bool(item7_3y_ago)
    return item1a or "", item7_latest or "", item7_3y_ago, has_comparison