Automating the New York DOS Entity Search With a Headless Fallback #

from __future__ import annotations

import asyncio
import hashlib
import json
import logging
from dataclasses import dataclass
from datetime import datetime, timezone
from enum import Enum

from playwright.async_api import (
    Page,
    TimeoutError as PlaywrightTimeout,
    async_playwright,
)

# Structured JSON logging — every line is a parseable audit/observability event.
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("ingestion.ny_dos")

# NY exposes a public inquiry UI but no stable JSON standing contract; the biennial
# Statement under N.Y. BCL s 408 is verified by reading this client-rendered grid.
NY_DOS_SEARCH_URL = "https://apps.dos.ny.gov/publicInquiry/"


class SearchOutcome(str, Enum):
    RESOLVED = "resolved"      # grid rendered and a status cell was read
    NOT_FOUND = "not_found"    # grid rendered with zero matching rows
    DEGRADED = "degraded"      # grid never rendered -> serve cached snapshot, flag stale
    CHALLENGED = "challenged"  # anti-bot boundary -> hard stop, escalate to human review


@dataclass(frozen=True)
class EntityStanding:
    entity_name: str
    dos_id: str | None
    status: str | None
    outcome: SearchOutcome
    retrieved_utc: str

    @property
    def evidence_sha256(self) -> str:
        # Hash the canonical record so a stored standing read is tamper-evident.
        body = json.dumps(
            {"name": self.entity_name, "dos_id": self.dos_id, "status": self.status,
             "outcome": self.outcome.value, "retrieved": self.retrieved_utc},
            sort_keys=True,
        ).encode("utf-8")
        return hashlib.sha256(body).hexdigest()


class CaptchaEncountered(RuntimeError):
    """A NY DOS anti-bot challenge — a terminal boundary; escalate, never retry."""


class NyDosSearchPage:
    """Page object for the NY DOS entity search: one method per interaction step."""

    NAME_INPUT = "input#SearchEntityName"        # the entity-name text field
    SEARCH_BUTTON = "button#SearchButton"        # submit control
    RESULTS_GRID = "table#SearchGrid tbody tr"   # JS-rendered result rows
    STATUS_CELL = "td[data-col='EntityStatus']"  # standing cell within a row
    DOS_ID_CELL = "td[data-col='DosId']"
    CHALLENGE = "iframe[src*='captcha'], #challenge-running"

    def __init__(self, page: Page) -> None:
        self._page = page

    async def open(self) -> None:
        # domcontentloaded, not networkidle: the grid is painted later via XHR.
        await self._page.goto(NY_DOS_SEARCH_URL, wait_until="domcontentloaded", timeout=20_000)
        await self._guard_challenge()

    async def query(self, entity_name: str) -> None:
        await self._page.fill(self.NAME_INPUT, entity_name)  # deterministic form fill
        await self._page.click(self.SEARCH_BUTTON)           # fires the results XHR

    async def await_results(self, timeout_ms: int = 12_000) -> bool:
        """Wait on the JS-rendered grid itself, never a timer. False => never rendered."""
        try:
            await self._page.wait_for_selector(self.RESULTS_GRID, timeout=timeout_ms)
            return True
        except PlaywrightTimeout:
            return False

    async def extract_first_row(self) -> tuple[str | None, str | None]:
        row = self._page.locator(self.RESULTS_GRID).first
        status = (await row.locator(self.STATUS_CELL).inner_text()).strip() or None
        dos_id = (await row.locator(self.DOS_ID_CELL).inner_text()).strip() or None
        return dos_id, status

    async def _guard_challenge(self) -> None:
        if await self._page.query_selector(self.CHALLENGE) is not None:
            raise CaptchaEncountered(self._page.url)


async def search_ny_dos(
    entity_name: str, cached: EntityStanding | None = None
) -> EntityStanding:
    """API-less NY DOS standing read with a headless fallback and graceful degradation."""
    now = datetime.now(timezone.utc).isoformat()
    async with async_playwright() as pw:
        browser = await pw.chromium.launch(
            headless=True,
            args=["--no-sandbox", "--disable-dev-shm-usage", "--disable-gpu"],
        )
        # Fresh, isolated context so no prior entity's cookies leak into this read.
        context = await browser.new_context(ignore_https_errors=False)
        page = await context.new_page()
        search = NyDosSearchPage(page)
        try:
            await search.open()
            await search.query(entity_name)
            if not await search.await_results():
                # DEGRADE GRACEFULLY: the grid never rendered. Do not fabricate a
                # status — serve the last cached snapshot flagged stale, or a bare
                # DEGRADED record the orchestrator escalates when no cache exists.
                logger.warning(json.dumps({
                    "event": "ny_dos_grid_timeout", "entity": entity_name,
                    "fell_back_to_cache": cached is not None,
                }))
                return cached or EntityStanding(
                    entity_name, None, None, SearchOutcome.DEGRADED, now
                )
            dos_id, status = await search.extract_first_row()
            outcome = SearchOutcome.RESOLVED if status else SearchOutcome.NOT_FOUND
            result = EntityStanding(entity_name, dos_id, status, outcome, now)
            logger.info(json.dumps({
                "event": "ny_dos_resolved", "entity": entity_name, "dos_id": dos_id,
                "status": status, "outcome": outcome.value,
                "evidence": result.evidence_sha256[:12],
            }))
            return result
        except CaptchaEncountered as exc:
            # Hard block: BCL s 408 standing is not worth crossing an access control.
            logger.warning(json.dumps({
                "event": "ny_dos_challenge", "entity": entity_name,
                "url": str(exc), "classification": "CAPTCHA_DETECTED",
            }))
            return EntityStanding(entity_name, None, None, SearchOutcome.CHALLENGED, now)
        finally:
            await context.close()  # never orphan a Chromium process


if __name__ == "__main__":
    standing = asyncio.run(search_ny_dos("ACME HOLDINGS LLC"))
    print(standing.outcome.value, standing.status, standing.evidence_sha256[:12])