diff --git a/openfoia/cli.py b/openfoia/cli.py index 49e79b9..89903fa 100644 --- a/openfoia/cli.py +++ b/openfoia/cli.py @@ -4142,6 +4142,103 @@ def records_search( rprint(f"\n[dim]Showing {len(entities)} of {result.total_results} results.[/dim]") +@records_app.command("filings") +def records_filings( + ticker_or_cik: str = typer.Argument(..., help="Company ticker or SEC CIK"), + since: str | None = typer.Option(None, "--since", help="Earliest filing date (YYYY-MM-DD)"), + forms: str | None = typer.Option(None, "--forms", help="Comma-separated forms, e.g. 10-K,10-Q"), + limit: int = typer.Option(25, "--limit", "-n", help="Maximum filings to display"), + tor: bool | None = typer.Option(None, "--tor/--no-tor", help="Route SEC requests through Tor"), + yes: bool = typer.Option(False, "--yes", help="Confirm sending the query to SEC"), +): + """List a company's SEC filings in reverse chronological order. + + The ticker or CIK is sent to SEC. This command never downloads filing + documents; it only retrieves SEC metadata and archive indexes. + """ + from datetime import date + + from .config import load_config + from .net import describe_egress + from .records.sec_edgar import SECEdgarAdapter + + if limit < 1: + rprint("[red]--limit must be at least 1.[/red]") + raise typer.Exit(2) + if since: + try: + date.fromisoformat(since) + except ValueError: + rprint("[red]--since must be an ISO date in YYYY-MM-DD format.[/red]") + raise typer.Exit(2) from None + try: + import re + + SECEdgarAdapter._validate_forms(forms) + if ticker_or_cik.strip().isdigit(): + SECEdgarAdapter.normalize_cik(ticker_or_cik) + elif not re.fullmatch(r"[A-Za-z][A-Za-z0-9.-]{0,9}", ticker_or_cik.strip()): + raise ValueError("ticker must be 1-10 letters, digits, periods, or hyphens") + except ValueError as exc: + rprint(f"[red]Invalid filing query: {exc}[/red]") + raise typer.Exit(2) from None + + cfg = load_config() + policy = _egress_policy_from(cfg, tor=tor) + _check_tor_or_exit(policy) + egress_info = describe_egress(policy) + rprint("\n[yellow]WARNING: This will send the ticker/CIK and filing query to SEC.[/yellow]") + if policy.is_tor: + rprint( + f"[cyan]Egress: Tor — SEC will not see your real IP " + f"(stream isolation: {egress_info['stream_isolation']}).[/cyan]" + ) + else: + rprint("[yellow]Egress: direct — SEC will see your real IP.[/yellow]") + rprint( + "[dim]Tor hides who is asking, not the ticker/CIK or query contents. " + "No filing documents will be downloaded.[/dim]" + ) + if not yes and not typer.confirm("Send this query to SEC?"): + rprint("[green]Aborted. Nothing left your machine.[/green]") + raise typer.Exit(0) + + import asyncio + + adapter = SECEdgarAdapter(egress=policy) + try: + result = asyncio.run(adapter.filings(ticker_or_cik, since=since, forms=forms)) + except ValueError as exc: + rprint(f"[red]Invalid filing query: {exc}[/red]") + raise typer.Exit(2) from None + if result.error: + rprint(f"[red]SEC source error: {result.error}[/red]") + raise typer.Exit(1) + if not result.entities: + rprint("[yellow]No SEC filings matched the requested filters.[/yellow]") + return + table = Table() + table.add_column("Date", width=12) + table.add_column("Form", width=10) + table.add_column("CIK", width=12) + table.add_column("Accession", width=22) + table.add_column("URL", max_width=70) + for entity in result.entities[:limit]: + table.add_row( + entity.extra_data.get("filing_date", "-"), + entity.extra_data.get("filing_type", "-"), + entity.identifiers.get("cik", "-"), + entity.identifiers.get("accession_number", "-"), + entity.source_url or "-", + ) + rprint(f"\n[bold]SEC filings for {ticker_or_cik}[/bold]") + console.print(table) + rprint( + f"\n[dim]Showing {min(limit, len(result.entities))} of " + f"{len(result.entities)} filings.[/dim]" + ) + + @records_app.command("fetch") def records_fetch( doc_id: str = typer.Argument(..., help="Document ID to fetch"), diff --git a/openfoia/records/sec_edgar.py b/openfoia/records/sec_edgar.py index c7ab652..229c7ea 100644 --- a/openfoia/records/sec_edgar.py +++ b/openfoia/records/sec_edgar.py @@ -6,12 +6,16 @@ from __future__ import annotations +import re +from datetime import date from typing import Any from .base import AdapterRequestError, RecordAdapter, RecordEntity, SearchResult EFTS_BASE = "https://efts.sec.gov/LATEST/search-index" EDGAR_FILING_BASE = "https://www.sec.gov/Archives/edgar/data" +SEC_TICKERS_URL = "https://www.sec.gov/files/company_tickers.json" +SEC_SUBMISSIONS_BASE = "https://data.sec.gov/submissions" #: SEC requires a descriptive User-Agent. Keep it generic and overridable — #: announcing "OpenFOIA" tells a government endpoint that the requester is @@ -31,6 +35,208 @@ class SECEdgarAdapter(RecordAdapter): source_name = "sec" + @staticmethod + def normalize_cik(value: str) -> str: + """Validate and return a CIK in SEC's ten-digit representation.""" + value = value.strip() + if not re.fullmatch(r"\d{1,10}", value): + raise ValueError("CIK must contain 1-10 decimal digits") + return value.zfill(10) + + async def resolve_ticker(self, ticker: str) -> tuple[str, str]: + """Resolve a ticker to ``(ten_digit_cik, company_name)``.""" + ticker = ticker.strip().upper() + if not re.fullmatch(r"[A-Z][A-Z0-9.-]{0,9}", ticker): + raise ValueError("ticker must be 1-10 letters, digits, periods, or hyphens") + data = await self._request(SEC_TICKERS_URL, headers=_edgar_headers()) + if not isinstance(data, dict): + raise AdapterRequestError("SEC ticker response was not a JSON object") + rows = data.values() + for row in rows: + if not isinstance(row, dict): + raise AdapterRequestError("SEC ticker response contained a non-object row") + if str(row.get("ticker", "")).upper() != ticker: + continue + try: + cik = self.normalize_cik(str(row.get("cik_str", ""))) + except ValueError as exc: + raise AdapterRequestError("SEC ticker response contained an invalid CIK") from exc + return cik, str(row.get("title") or ticker) + raise ValueError(f"ticker '{ticker}' was not found in SEC company_tickers.json") + + @staticmethod + def _validate_forms(forms: str | None) -> set[str] | None: + if not forms: + return None + values = {part.strip().upper() for part in forms.split(",")} + if not values or any( + not re.fullmatch(r"[A-Z0-9](?:[A-Z0-9./ -]*[A-Z0-9])?", value) for value in values + ): + raise ValueError("forms must be a comma-separated list such as 10-K,10-Q,8-K") + return values + + @staticmethod + def _filing_rows(block: dict[str, Any], *, allow_empty: bool = True) -> list[dict[str, Any]]: + """Validate and transpose SEC's column-oriented metadata into rows.""" + if not block: + if not allow_empty: + raise AdapterRequestError("SEC archive filing metadata block was empty") + return [] + required = ("accessionNumber", "filingDate", "form") + if any(key not in block or not isinstance(block[key], list) for key in required): + raise AdapterRequestError("SEC filing metadata block is missing a required column") + length = len(block[required[0]]) + if length == 0 and not allow_empty: + raise AdapterRequestError("SEC archive filing metadata block was empty") + if any(not isinstance(value, list) or len(value) != length for value in block.values()): + raise AdapterRequestError("SEC filing metadata columns have unequal lengths") + return [ + {key: value[index] for key, value in block.items() if isinstance(value, list)} + for index in range(length) + ] + + @staticmethod + def _filing_entity( + filing: dict[str, Any], *, cik: str, company_name: str + ) -> RecordEntity | None: + if not isinstance(filing, dict): + raise AdapterRequestError("SEC filing row was not an object") + accession = ( + filing["accessionNumber"] if "accessionNumber" in filing else filing.get("accession_no") + ) + filing_date = filing["filingDate"] if "filingDate" in filing else filing.get("file_date") + form = filing["form"] if "form" in filing else filing.get("form_type") + if ( + not isinstance(accession, str) + or not isinstance(filing_date, str) + or not isinstance(form, str) + ): + raise AdapterRequestError("SEC filing row contained non-string metadata") + if ( + not re.fullmatch(r"\d{10}-\d{2}-\d{6}", accession) + or not re.fullmatch(r"\d{4}-\d{2}-\d{2}", filing_date) + or not re.fullmatch(r"[A-Z0-9](?:[A-Z0-9./ -]*[A-Z0-9])?", form.upper()) + ): + raise AdapterRequestError("SEC filing row contained invalid metadata") + try: + date.fromisoformat(filing_date) + except ValueError as exc: + raise AdapterRequestError("SEC filing row contained an invalid filing date") from exc + primary_value = filing.get("primaryDocument") + if primary_value is not None and not isinstance(primary_value, str): + raise AdapterRequestError("SEC filing row contained a non-string primary document") + primary = primary_value or "" + # SEC metadata is remote input; only a plain archive basename may be + # interpolated into a URL. Suspicious values use the safe index URL. + safe_primary = ( + primary + if primary not in {".", ".."} and re.fullmatch(r"[A-Za-z0-9._-]+", primary) + else "" + ) + clean_accession = accession.replace("-", "") + numeric_cik = str(int(cik)) + if safe_primary: + url = f"{EDGAR_FILING_BASE}/{numeric_cik}/{clean_accession}/{safe_primary}" + else: + url = f"{EDGAR_FILING_BASE}/{numeric_cik}/{clean_accession}/{accession}-index.htm" + return RecordEntity( + entity_type="ORGANIZATION", + name=company_name, + source="sec", + source_url=url, + jurisdiction="us_federal", + identifiers={"cik": cik, "accession_number": accession}, + extra_data={ + "filing_type": form, + "filing_date": filing_date, + "primary_document": primary, + }, + ) + + async def filings( + self, + company: str, + *, + since: str | None = None, + forms: str | None = None, + ) -> SearchResult: + """Return a company's complete recent and archived filing history.""" + if since: + try: + since_date = date.fromisoformat(since) + except ValueError as exc: + raise ValueError("since must be an ISO date in YYYY-MM-DD format") from exc + else: + since_date = None + form_filter = self._validate_forms(forms) + company = company.strip() + if not company: + raise ValueError("ticker or CIK is required") + + try: + if re.fullmatch(r"\d{1,10}", company): + cik = self.normalize_cik(company) + company_name = cik + else: + cik, company_name = await self.resolve_ticker(company) + root_url = f"{SEC_SUBMISSIONS_BASE}/CIK{cik}.json" + root = await self._request(root_url, headers=_edgar_headers()) + if not isinstance(root, dict): + raise AdapterRequestError("SEC submissions response was not a JSON object") + submissions = root.get("filings") + if not isinstance(submissions, dict): + raise AdapterRequestError("SEC submissions filings block was not a JSON object") + if "recent" not in submissions or "files" not in submissions: + raise AdapterRequestError( + "SEC submissions filings block was missing recent or files" + ) + filings: list[dict[str, Any]] = [] + recent = submissions["recent"] + if not isinstance(recent, dict): + raise AdapterRequestError("SEC submissions recent block was not a JSON object") + if isinstance(recent, dict): + filings.extend(self._filing_rows(recent)) + archive_files = submissions["files"] + if not isinstance(archive_files, list): + raise AdapterRequestError("SEC submissions files block was not a list") + archive_pattern = re.compile(rf"CIK{re.escape(cik)}-submissions-\d{{3}}\.json") + for archive in archive_files: + if not isinstance(archive, dict) or not isinstance(archive.get("name"), str): + raise AdapterRequestError("SEC submissions archive entry was malformed") + name = archive["name"] + if not archive_pattern.fullmatch(name): + raise AdapterRequestError("SEC submissions archive name was malformed") + archived = await self._request( + f"{SEC_SUBMISSIONS_BASE}/{name}", headers=_edgar_headers() + ) + if not isinstance(archived, dict): + raise AdapterRequestError(f"SEC archive {name} was not a JSON object") + block = archived.get("filings", archived) + if not isinstance(block, dict): + raise AdapterRequestError(f"SEC archive {name} filings block was invalid") + filings.extend(self._filing_rows(block, allow_empty=False)) + if isinstance(root.get("name"), str) and root["name"]: + company_name = root["name"] + entities = [] + for filing in filings: + entity = self._filing_entity(filing, cik=cik, company_name=company_name) + filing_date = entity.extra_data["filing_date"] + if since_date and filing_date < since_date.isoformat(): + continue + if form_filter and entity.extra_data["filing_type"].upper() not in form_filter: + continue + entities.append(entity) + entities.sort(key=lambda entity: entity.extra_data["filing_date"], reverse=True) + return SearchResult( + source=self.source_name, + query=company, + total_results=len(entities), + entities=entities, + raw_response=root, + ) + except AdapterRequestError as exc: + return self._failed(company, str(exc)) + async def search(self, query: str, **kwargs: Any) -> SearchResult: """Search SEC EDGAR filings by keyword. diff --git a/tests/test_sec_filings.py b/tests/test_sec_filings.py new file mode 100644 index 0000000..6e14519 --- /dev/null +++ b/tests/test_sec_filings.py @@ -0,0 +1,300 @@ +"""Offline tests for SEC company filing history.""" + +from __future__ import annotations + +import asyncio + +import pytest +from typer.testing import CliRunner + +from openfoia.cli import app +from openfoia.config import OpenFOIAConfig +from openfoia.net import EgressMode, EgressPolicy +from openfoia.records.base import AdapterRequestError, SearchResult +from openfoia.records.sec_edgar import ( + SEC_SUBMISSIONS_BASE, + SEC_TICKERS_URL, + SECEdgarAdapter, +) + + +def test_normalize_cik_and_reject_invalid_values() -> None: + assert SECEdgarAdapter.normalize_cik("123") == "0000000123" + with pytest.raises(ValueError, match="CIK"): + SECEdgarAdapter.normalize_cik("12345678901") + assert SECEdgarAdapter._validate_forms("10-K/A, SC 13D") == {"10-K/A", "SC 13D"} + with pytest.raises(ValueError): + SECEdgarAdapter._validate_forms("10-K,../bad") + + +def test_filings_resolves_ticker_aggregates_archives_and_filters() -> None: + class FakeAdapter(SECEdgarAdapter): + def __init__(self) -> None: + super().__init__(egress=EgressPolicy(mode=EgressMode.TOR)) + self.calls: list[tuple[str, dict]] = [] + + async def _request(self, url: str, **kwargs): # type: ignore[no-untyped-def] + self.calls.append((url, kwargs)) + if url == SEC_TICKERS_URL: + return {"0": {"ticker": "ABC", "cik_str": 123, "title": "ABC Corp"}} + if url.endswith("CIK0000000123.json"): + return { + "name": "ABC Corp", + "filings": { + "recent": { + "accessionNumber": [ + "0000000001-02-000002", + "0000000001-01-000001", + ], + "filingDate": ["2025-01-02", "2024-01-01"], + "form": ["10-K", "8-K"], + "primaryDocument": ["annual.htm", "event.htm"], + }, + "files": [{"name": "CIK0000000123-submissions-001.json"}], + }, + } + return { + "accessionNumber": ["0000000001-01-000003"], + "filingDate": ["2023-01-01"], + "form": ["10-K"], + "primaryDocument": ["old.htm"], + } + + adapter = FakeAdapter() + result = asyncio.run(adapter.filings("abc")) + + assert [e.extra_data["filing_date"] for e in result.entities] == [ + "2025-01-02", + "2024-01-01", + "2023-01-01", + ] + assert result.entities[0].source_url.endswith("/123/000000000102000002/annual.htm") + assert result.entities[-1].source_url.endswith("/123/000000000101000003/old.htm") + assert all(kwargs["headers"]["User-Agent"] for _, kwargs in adapter.calls) + assert all("_egress" not in kwargs for _, kwargs in adapter.calls) + assert adapter.calls[0][0] == SEC_TICKERS_URL + assert adapter.calls[1][0] == f"{SEC_SUBMISSIONS_BASE}/CIK0000000123.json" + + +def test_suspicious_primary_document_falls_back_to_index() -> None: + entity = SECEdgarAdapter._filing_entity( + { + "accessionNumber": "0000000001-02-000001", + "filingDate": "2025-01-02", + "form": "10-K", + "primaryDocument": "../secret.txt", + }, + cik="0000000123", + company_name="ABC", + ) + assert entity is not None + assert entity.source_url.endswith("/123/000000000102000001/0000000001-02-000001-index.htm") + + +@pytest.mark.parametrize( + "filing", + [ + {"accessionNumber": "bad", "filingDate": "2025-01-02", "form": "10-K"}, + {"accessionNumber": "0000000001-02-000001", "filingDate": "2025-99-99", "form": "10-K"}, + ], +) +def test_invalid_filing_metadata_is_skipped(filing) -> None: # type: ignore[no-untyped-def] + with pytest.raises(AdapterRequestError): + SECEdgarAdapter._filing_entity(filing, cik="0000000123", company_name="ABC") + + +@pytest.mark.parametrize( + "filing", + [ + { + "accessionNumber": "0000000001-02-000001", + "filingDate": "2025-01-02", + "form": ["10-K"], + }, + { + "accessionNumber": "0000000001-02-000001", + "filingDate": "2025-01-02", + "form": "10-K", + "primaryDocument": ["safe.htm"], + }, + ], +) +def test_non_string_remote_metadata_is_source_error(filing) -> None: # type: ignore[no-untyped-def] + with pytest.raises(AdapterRequestError): + SECEdgarAdapter._filing_entity(filing, cik="0000000123", company_name="ABC") + + +def test_unequal_optional_column_is_source_error() -> None: + with pytest.raises(AdapterRequestError): + SECEdgarAdapter._filing_rows( + { + "accessionNumber": ["0000000001-02-000001"], + "filingDate": ["2025-01-02"], + "form": ["10-K"], + "optional": [], + } + ) + + +def test_dot_primary_document_falls_back_to_index() -> None: + entity = SECEdgarAdapter._filing_entity( + { + "accessionNumber": "0000000001-02-000001", + "filingDate": "2025-01-02", + "form": "10-K", + "primaryDocument": "..", + }, + cik="0000000123", + company_name="ABC", + ) + assert entity is not None + assert entity.source_url.endswith("-index.htm") + + +def test_source_failure_is_not_reported_as_no_results() -> None: + class FailedAdapter(SECEdgarAdapter): + async def _request(self, url: str, **kwargs): # type: ignore[no-untyped-def] + raise AdapterRequestError("offline") + + result = asyncio.run(FailedAdapter().filings("ABC")) + assert result.entities == [] + assert result.error == "offline" + + +@pytest.mark.parametrize( + "malformed", + [ + None, + {}, + {"filings": {}}, + {"filings": []}, + {"accessionNumber": [], "filingDate": [], "form": []}, + ], +) +def test_malformed_archive_json_is_a_source_failure(malformed) -> None: # type: ignore[no-untyped-def] + class MalformedAdapter(SECEdgarAdapter): + async def _request(self, url: str, **kwargs): # type: ignore[no-untyped-def] + if url == SEC_TICKERS_URL: + return {"0": {"ticker": "ABC", "cik_str": 123, "title": "ABC Corp"}} + if url.endswith("CIK0000000123.json"): + return { + "filings": { + "recent": {}, + "files": [{"name": "CIK0000000123-submissions-001.json"}], + } + } + return malformed + + result = asyncio.run(MalformedAdapter().filings("ABC")) + assert result.entities == [] + assert result.error is not None + + +def test_malformed_root_json_is_a_source_failure() -> None: + class MalformedAdapter(SECEdgarAdapter): + async def _request(self, url: str, **kwargs): # type: ignore[no-untyped-def] + if url == SEC_TICKERS_URL: + return {"0": {"ticker": "ABC", "cik_str": 123, "title": "ABC Corp"}} + return [] + + result = asyncio.run(MalformedAdapter().filings("ABC")) + assert result.entities == [] + assert result.error is not None + + +@pytest.mark.parametrize( + "root", + [ + {}, + {"filings": {}}, + {"filings": {"recent": {}}}, + {"filings": {"recent": [], "files": []}}, + { + "filings": { + "recent": { + "accessionNumber": ["0000000001-02-000001"], + "filingDate": ["2025-01-02"], + }, + "files": [], + } + }, + { + "filings": { + "recent": { + "accessionNumber": ["0000000001-02-000001"], + "filingDate": ["2025-01-02"], + "form": ["10-K"], + "primaryDocument": [], + }, + "files": [], + } + }, + ], +) +def test_submission_schema_drift_is_a_source_failure(root) -> None: # type: ignore[no-untyped-def] + class SchemaAdapter(SECEdgarAdapter): + async def _request(self, url: str, **kwargs): # type: ignore[no-untyped-def] + return root + + result = asyncio.run(SchemaAdapter().filings("123")) + assert result.entities == [] + assert result.error is not None + + +def test_invalid_row_is_a_source_failure_before_filters() -> None: + class InvalidRowAdapter(SECEdgarAdapter): + async def _request(self, url: str, **kwargs): # type: ignore[no-untyped-def] + return { + "filings": { + "recent": { + "accessionNumber": ["not-an-accession"], + "filingDate": ["2020-01-01"], + "form": ["10-K"], + }, + "files": [], + } + } + + result = asyncio.run(InvalidRowAdapter().filings("123", since="2025-01-01", forms="10-K")) + assert result.entities == [] + assert result.error is not None + + +def test_cli_decline_does_not_construct_adapter(monkeypatch) -> None: # type: ignore[no-untyped-def] + runner = CliRunner() + monkeypatch.setattr("openfoia.config.load_config", lambda: OpenFOIAConfig()) + monkeypatch.setattr("openfoia.cli._check_tor_or_exit", lambda policy: None) + monkeypatch.setattr("typer.confirm", lambda prompt: False) + + class ExplodingAdapter(SECEdgarAdapter): + def __init__(self, **kwargs): + raise AssertionError("adapter must not be constructed after decline") + + monkeypatch.setattr("openfoia.records.sec_edgar.SECEdgarAdapter", ExplodingAdapter) + result = runner.invoke(app, ["records", "filings", "ABC"]) + assert result.exit_code == 0 + assert "Nothing left your machine" in result.output + + +def test_cli_yes_forwards_policy_and_reports_source_error(monkeypatch) -> None: # type: ignore[no-untyped-def] + runner = CliRunner() + config = OpenFOIAConfig() + config.network.tor = True + monkeypatch.setattr("openfoia.config.load_config", lambda: config) + monkeypatch.setattr("openfoia.cli._check_tor_or_exit", lambda policy: None) + seen: dict[str, object] = {} + + class FakeAdapter(SECEdgarAdapter): + def __init__(self, **kwargs): + seen.update(kwargs) + + async def filings(self, company, **kwargs): # type: ignore[no-untyped-def] + return SearchResult( + source="sec", query=company, total_results=0, entities=[], error="offline" + ) + + monkeypatch.setattr("openfoia.records.sec_edgar.SECEdgarAdapter", FakeAdapter) + result = runner.invoke(app, ["records", "filings", "ABC", "--yes"]) + assert result.exit_code == 1 + assert "SEC source error: offline" in result.output + assert getattr(seen["egress"], "is_tor", False) is True