How to Map LLC vs C-Corp Filing Requirements Across 50 States #

import hashlib
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from itertools import islice
from typing import Any, Dict, Generator, List, Optional

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Structured, JSON-friendly logging so audit context survives log shipping.
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%SZ",
)
logger = logging.getLogger("compliance.engine")


class EntityType(str, Enum):
    LLC = "LLC"
    CCORP = "C-CORP"


@dataclass(frozen=True)
class ComplianceRule:
    # Obligation is keyed on (state, type, effective window) — never on state alone.
    state_code: str
    entity_type: EntityType
    report_required: bool          # Delaware LLCs are False; C-Corps are True.
    deadline_month: int
    deadline_day: int
    fee_formula: str               # e.g. "flat_300" vs "assumed_par_value".
    version: str                   # Bumped on every statutory amendment.
    effective_from: datetime


@dataclass
class AuditEntry:
    entity_id: str
    state_code: str
    entity_type: EntityType
    rule_version: str
    evaluated_at: datetime
    status: str
    previous_hash: str
    current_hash: str = field(init=False)

    def __post_init__(self) -> None:
        # Forward-chain each record into the previous hash for tamper evidence.
        payload = (
            f"{self.entity_id}:{self.state_code}:{self.entity_type.value}:"
            f"{self.rule_version}:{self.status}:{self.previous_hash}"
        )
        self.current_hash = hashlib.sha256(payload.encode()).hexdigest()


class PortalClient:
    """Circuit-breaking client for a single Secretary of State portal."""

    def __init__(self, base_url: str, timeout: float = 10.0, threshold: int = 5):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.mount("https://", HTTPAdapter(max_retries=Retry(
            total=3, backoff_factor=0.5, status_forcelist=[429, 500, 502, 503, 504]
        )))
        self.timeout = timeout
        self._circuit_open = False
        self._failure_count = 0
        self._threshold = threshold

    def _check_circuit(self) -> bool:
        if self._circuit_open:
            logger.warning("Circuit open for %s; using cached compliance state.", self.base_url)
            return False
        return True

    def _record_failure(self) -> None:
        self._failure_count += 1
        if self._failure_count >= self._threshold:
            self._circuit_open = True
            logger.error("Circuit breaker tripped for %s", self.base_url)

    def _reset_circuit(self) -> None:
        self._failure_count = 0
        self._circuit_open = False

    def fetch_entity_status(
        self, entity_id: str, auth_token: Optional[str] = None
    ) -> Dict[str, Any]:
        if not self._check_circuit():
            # Fallback path: never block the run on a dead portal.
            return {"status": "cached", "entity_id": entity_id, "fallback": True}

        headers = {"Authorization": f"Bearer {auth_token}"} if auth_token else {}
        try:
            resp = self.session.get(
                f"{self.base_url}/api/v1/entities/{entity_id}",
                headers=headers,
                timeout=self.timeout,
            )
            resp.raise_for_status()
            self._reset_circuit()
            return resp.json()
        except requests.exceptions.RequestException as exc:
            self._record_failure()
            logger.error("Portal fetch failed for %s: %s", entity_id, exc)
            return {"status": "error", "entity_id": entity_id, "fallback": True, "error": str(exc)}


class ComplianceRuleEngine:
    def __init__(self, rules: List[ComplianceRule]):
        # Compound key guarantees the LLC and C-Corp rules never collide.
        self.rules: Dict[tuple[str, EntityType], ComplianceRule] = {
            (r.state_code, r.entity_type): r for r in rules
        }
        self.audit_trail: List[AuditEntry] = []
        self._last_hash = "genesis"

    def resolve(self, state_code: str, entity_type: EntityType) -> Optional[ComplianceRule]:
        return self.rules.get((state_code, entity_type))

    def evaluate_chunk(
        self, entities: List[Dict[str, Any]], portal: PortalClient
    ) -> Generator[AuditEntry, None, None]:
        for entity in entities:
            eid = entity["id"]
            state = entity["state_code"]
            etype = EntityType(entity["type"])
            rule = self.resolve(state, etype)

            if rule is None:
                # Halt rather than guess: an unmapped (state, type) is a data gap.
                logger.warning("No rule for %s/%s on %s; sent to review.", state, etype.value, eid)
                continue

            portal_data = portal.fetch_entity_status(eid)
            status = "compliant" if portal_data.get("status") == "active" else "pending_review"

            entry = AuditEntry(
                entity_id=eid,
                state_code=state,
                entity_type=etype,
                rule_version=rule.version,
                evaluated_at=datetime.now(timezone.utc),
                status=status,
                previous_hash=self._last_hash,
            )
            self._last_hash = entry.current_hash
            self.audit_trail.append(entry)
            logger.info(
                "Evaluated %s | %s/%s | %s | hash=%s",
                eid, state, etype.value, status, entry.current_hash,
            )
            yield entry


def stream_entities(
    entity_source: List[Dict[str, Any]], chunk_size: int = 500
) -> Generator[List[Dict[str, Any]], None, None]:
    # islice keeps memory bounded regardless of portfolio size.
    iterator = iter(entity_source)
    while True:
        chunk = list(islice(iterator, chunk_size))
        if not chunk:
            break
        yield chunk


if __name__ == "__main__":
    active_rules = [
        ComplianceRule("DE", EntityType.LLC, False, 6, 1, "flat_300", "v2.1",
                       datetime(2024, 1, 1, tzinfo=timezone.utc)),
        ComplianceRule("DE", EntityType.CCORP, True, 3, 1, "assumed_par_value", "v2.1",
                       datetime(2024, 1, 1, tzinfo=timezone.utc)),
    ]
    engine = ComplianceRuleEngine(active_rules)
    client = PortalClient("https://portal.example-state.gov")

    sample_entities = [
        {"id": "ENT-001", "state_code": "DE", "type": "LLC"},
        {"id": "ENT-002", "state_code": "DE", "type": "C-CORP"},
    ]

    for chunk in stream_entities(sample_entities, chunk_size=500):
        for audit in engine.evaluate_chunk(chunk, client):
            pass  # Persist each entry to the immutable NDJSON sink here.