"""Small standard-library client for the FilingSift REST API v1.""" from __future__ import annotations import json import re from typing import Any, Mapping from urllib.error import HTTPError, URLError from urllib.parse import quote, urlencode, urlsplit from urllib.request import Request, urlopen def _validated_base_url(value: str) -> str: message = "base_url must be an HTTPS URL or exact localhost/127.0.0.1 HTTP URL, without credentials, query, or fragment" if (not isinstance(value, str) or re.search(r"[\s\\\x00-\x1f\x7f]", value) or not re.fullmatch(r"https?://(?:\[[^\]]+\]|[^/:?#@]+)(?::[0-9]+)?(?:/[^?#]*)?", value, re.IGNORECASE)): raise ValueError(message) try: parsed = urlsplit(value) port = parsed.port # Access validates numeric port syntax and range. if (not parsed.hostname or parsed.username is not None or parsed.password is not None or re.search(r'[%<>"^|]', parsed.netloc) or (port is not None and port < 1) or (parsed.scheme == "http" and parsed.hostname not in ("localhost", "127.0.0.1"))): raise ValueError(message) except ValueError: raise ValueError(message) from None return value.rstrip("/") class FilingSiftError(RuntimeError): """A transport or structured FilingSift API error.""" def __init__( self, message: str, *, status: int | None = None, code: str = "request_failed", request_id: str | None = None, retryable: bool = False, body: Any = None, ) -> None: super().__init__(message) self.status = status self.code = code self.request_id = request_id self.retryable = retryable self.body = body class FilingSiftClient: """Call FilingSift v1 without changing the response schema.""" def __init__( self, api_key: str | None = None, *, base_url: str = "https://api.filingsift.com/api/v1", timeout: float = 15.0, ) -> None: validated_base_url = _validated_base_url(base_url) if timeout <= 0: raise ValueError("timeout must be positive") self.api_key = api_key self.base_url = validated_base_url self.timeout = timeout def request( self, method: str, path: str, *, query: Mapping[str, Any] | None = None, json_body: Mapping[str, Any] | None = None, accept: str = "application/json", ) -> Any: """Return the server response without reshaping filing-derived facts.""" normalized_path = "/" + path.lstrip("/") parameters = {key: value for key, value in (query or {}).items() if value is not None} url = f"{self.base_url}{normalized_path}" if parameters: url = f"{url}?{urlencode(parameters, doseq=True)}" headers = {"Accept": accept, "User-Agent": "filingsift-python/0.2.0"} if self.api_key: headers["Authorization"] = f"Bearer {self.api_key}" payload = None if json_body is not None: payload = json.dumps(json_body).encode("utf-8") headers["Content-Type"] = "application/json" request = Request(url, data=payload, headers=headers, method=method.upper()) try: with urlopen(request, timeout=self.timeout) as response: if response.status in (204, 205): return None content = response.read() content_type = response.headers.get_content_type() if response.headers.get("Content-Type") else "" except HTTPError as error: body = self._decode_error(error) details = body.get("error", {}) if isinstance(body, dict) else {} reason = details.get("reason", {}) if isinstance(details, dict) else {} message = details.get("message") if isinstance(details, dict) else None raise FilingSiftError( message or f"FilingSift returned HTTP {error.code}", status=error.code, code=details.get("code", "request_failed") if isinstance(details, dict) else "request_failed", request_id=(body.get("requestId") if isinstance(body, dict) else None) or error.headers.get("X-Request-Id"), retryable=bool(reason.get("retryable")) if isinstance(reason, dict) else False, body=body, ) from error except URLError as error: raise FilingSiftError(str(error.reason), code="network_error", retryable=True) from error if accept.startswith("text/") or content_type.startswith("text/"): return content.decode("utf-8") binary_types = {"application/pdf", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"} if accept in binary_types or content_type in binary_types: return content try: return json.loads(content) except (UnicodeDecodeError, json.JSONDecodeError) as error: raise FilingSiftError("FilingSift returned an invalid JSON response", code="invalid_response") from error @staticmethod def _decode_error(error: HTTPError) -> Any: try: return json.loads(error.read()) except (UnicodeDecodeError, json.JSONDecodeError): return None def get_company(self, ticker: str) -> dict[str, Any]: return self.request("GET", f"/companies/{quote(ticker.upper(), safe='')}") def list_comparisons(self, **filters: Any) -> dict[str, Any]: return self.request("GET", "/filing-comparisons", query=filters) def get_coverage(self) -> dict[str, Any]: return self.request("GET", "/coverage") def get_filing_reader(self, accession: str) -> dict[str, Any]: return self.request("GET", f"/reader/filings/{quote(accession, safe='')}") def compare_filing_reader(self, accession: str, other_accession: str) -> dict[str, Any]: return self.request("GET", f"/reader/filings/{quote(accession, safe='')}/compare/{quote(other_accession, safe='')}") def export_evidence_report(self, accession: str, compare_to: str | None = None) -> bytes: return self.request("GET", f"/reader/filings/{quote(accession, safe='')}/report.pdf", query={"compare_to": compare_to}, accept="application/pdf") def add_company(self, ticker: str) -> dict[str, Any]: return self.request("POST", "/account/companies", json_body={"ticker": ticker.upper()}) def retry_onboarding(self, ticker: str) -> dict[str, Any]: return self.request("POST", f"/account/companies/{quote(ticker.upper(), safe='')}/onboarding") def get_monitoring(self) -> dict[str, Any]: return self.request("GET", "/account/monitoring") def update_monitoring_preferences(self, preferences: Mapping[str, Any]) -> dict[str, Any]: return self.request("PATCH", "/account/monitoring/preferences", json_body=preferences) def send_monitoring_test(self) -> dict[str, Any]: return self.request("POST", "/account/monitoring/test") def list_monitoring_deliveries(self, limit: int = 50) -> dict[str, Any]: return self.request("GET", "/account/monitoring/deliveries", query={"limit": limit}) def retry_monitoring_delivery(self, delivery_id: str) -> dict[str, Any]: return self.request("POST", f"/account/monitoring/deliveries/{quote(delivery_id, safe='')}/retry") def search_filings(self, request: Mapping[str, Any]) -> dict[str, Any]: return self.request("POST", "/search", json_body=request) def search_within_filing(self, accession: str, request: Mapping[str, Any]) -> dict[str, Any]: return self.request("POST", f"/reader/filings/{quote(accession, safe='')}/search", json_body=request) def screen_financing(self, request: Mapping[str, Any]) -> dict[str, Any]: return self.request("POST", "/screens/financing", json_body=request) def compare_financing_matrix(self, request: Mapping[str, Any]) -> dict[str, Any]: return self.request("POST", "/screens/comparison-matrix", json_body=request) def get_financing_intelligence(self, ticker: str) -> dict[str, Any]: return self.request("GET", f"/intelligence/issuers/{quote(ticker.upper(), safe='')}/financing") def get_financing_timeline(self, ticker: str) -> dict[str, Any]: return self.request("GET", f"/intelligence/issuers/{quote(ticker.upper(), safe='')}/timeline") def get_financing_quality(self, ticker: str) -> dict[str, Any]: return self.request("GET", f"/intelligence/issuers/{quote(ticker.upper(), safe='')}/quality") def build_analyst_financing_table(self, request: Mapping[str, Any]) -> dict[str, Any]: return self.request("POST", "/analyst/tables/financing", json_body=request) def export_analyst_workbook(self, request: Mapping[str, Any]) -> bytes: return self.request( "POST", "/analyst/exports/xlsx", json_body=request, accept="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) def get_analyst_workbook_manifest(self, request: Mapping[str, Any]) -> dict[str, Any]: return self.request("POST", "/analyst/exports/xlsx/manifest", json_body=request) def get_analyst_modeling_vocabulary(self) -> dict[str, Any]: return self.request("GET", "/analyst/modeling/vocabulary") def resolve_analyst_formulas(self, request: Mapping[str, Any]) -> dict[str, Any]: return self.request("POST", "/analyst/formulas/resolve", json_body=request) def render_analyst_template(self, request: Mapping[str, Any]) -> dict[str, Any]: return self.request("POST", "/analyst/templates/render", json_body=request) def list_saved_searches(self) -> dict[str, Any]: return self.request("GET", "/account/saved-searches") def create_saved_search(self, request: Mapping[str, Any]) -> dict[str, Any]: return self.request("POST", "/account/saved-searches", json_body=request) def update_saved_search(self, saved_search_id: str, request: Mapping[str, Any]) -> dict[str, Any]: return self.request("PATCH", f"/account/saved-searches/{quote(saved_search_id, safe='')}", json_body=request) def delete_saved_search(self, saved_search_id: str) -> None: return self.request("DELETE", f"/account/saved-searches/{quote(saved_search_id, safe='')}") def get_developer_sandbox(self) -> dict[str, Any]: return self.request("GET", "/developer/sandbox") def get_status_history(self, days: int = 30) -> dict[str, Any]: return self.request("GET", "/status/history", query={"days": days}) def get_incidents(self) -> dict[str, Any]: return self.request("GET", "/incidents") def create_archive(self, kind: str, accessions: list[str]) -> dict[str, Any]: return self.request("POST", "/account/batch-jobs", json_body={"kind": kind, "items": accessions}) def list_archives(self) -> dict[str, Any]: return self.request("GET", "/account/batch-jobs") def get_archive(self, job_id: str) -> dict[str, Any]: return self.request("GET", f"/account/batch-jobs/{quote(job_id, safe='')}") def download_archive(self, job_id: str) -> dict[str, Any]: return self.request("GET", f"/account/batch-jobs/{quote(job_id, safe='')}/download") def get_parsed_html(self, accession: str) -> str: return self.request("GET", f"/filings/{quote(accession, safe='')}/parsed.html", accept="text/html") def get_parsed_text(self, accession: str) -> str: return self.request("GET", f"/filings/{quote(accession, safe='')}/parsed.txt", accept="text/plain") def get_parsed_tables(self, accession: str) -> dict[str, Any]: return self.request("GET", f"/filings/{quote(accession, safe='')}/tables.json") def get_xbrl_json(self, accession: str) -> dict[str, Any]: return self.request("GET", f"/filings/{quote(accession, safe='')}/xbrl.json") def get_clean_pdf(self, accession: str) -> bytes: return self.request("GET", f"/filings/{quote(accession, safe='')}/clean.pdf", accept="application/pdf") def stream_filings(self, *, cursor: str | None = None, cik: str | None = None, ticker: str | None = None, form: str | None = None): """Yield reconnectable filing.indexed SSE events without reshaping provenance.""" query = {key: value for key, value in {"cursor": cursor, "cik": cik, "ticker": ticker, "form": form}.items() if value is not None} url = f"{self.base_url}/filings/stream" if query: url = f"{url}?{urlencode(query)}" headers = {"Accept": "text/event-stream", "User-Agent": "filingsift-python/0.2.0"} if self.api_key: headers["Authorization"] = f"Bearer {self.api_key}" request = Request(url, headers=headers, method="GET") try: with urlopen(request, timeout=max(self.timeout, 30.0)) as response: event_name = None data_lines: list[str] = [] for raw_line in response: line = raw_line.decode("utf-8").rstrip("\r\n") if not line: if data_lines: try: payload = json.loads("\n".join(data_lines)) except json.JSONDecodeError as error: raise FilingSiftError("Filing stream returned invalid JSON", code="invalid_stream_event") from error if event_name == "error": raise FilingSiftError("Filing stream reported an error", code=payload.get("code", "stream_failed"), retryable=bool(payload.get("retryable")), body=payload) if event_name == "filing.indexed": yield payload event_name = None data_lines = [] elif line.startswith("event:"): event_name = line[6:].strip() elif line.startswith("data:"): data_lines.append(line[5:].lstrip()) except HTTPError as error: body = self._decode_error(error) details = body.get("error", {}) if isinstance(body, dict) else {} raise FilingSiftError(details.get("message") or f"FilingSift returned HTTP {error.code}", status=error.code, code=details.get("code", "request_failed"), body=body) from error except URLError as error: raise FilingSiftError(str(error.reason), code="network_error", retryable=True) from error def export_events_csv(self, **filters: Any) -> str: return self.request("GET", "/events.csv", query=filters, accept="text/csv")