Building a Fallback Routing System for Patent Dockets

import asyncio
import hashlib
import logging
import time
from datetime import datetime, timezone
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import httpx

# Configure structured audit logger
logging.basicConfig(
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    level=logging.INFO
)
logger = logging.getLogger("patent_docket_fallback")

@dataclass
class CircuitBreaker:
    failure_threshold: int = 3
    recovery_timeout: float = 30.0
    failures: int = 0
    last_failure_time: float = 0.0
    state: str = "closed"  # closed | open | half-open

    def record_failure(self) -> None:
        self.failures += 1
        self.last_failure_time = time.monotonic()
        if self.failures >= self.failure_threshold:
            self.state = "open"
            logger.warning("Circuit breaker opened after %d failures", self.failures)

    def reset(self) -> None:
        self.failures = 0
        self.state = "closed"

    def allow_request(self) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open" and (time.monotonic() - self.last_failure_time) > self.recovery_timeout:
            self.state = "half-open"
            return True
        return False

class DocketRoutingError(Exception):
    pass

class FallbackRouter:
    def __init__(self, endpoints: List[str]):
        self.endpoints = endpoints
        self.circuit_breakers = {url: CircuitBreaker() for url in endpoints}
        self.client = httpx.AsyncClient(timeout=3.0, follow_redirects=False)
        self.audit_log: List[Dict[str, Any]] = []

    def _generate_payload_hash(self, payload: Dict[str, Any]) -> str:
        canonical = str(sorted(payload.items()))
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

    async def _route_with_retry(self, url: str, payload: Dict[str, Any]) -> bool:
        breaker = self.circuit_breakers[url]
        if not breaker.allow_request():
            logger.info("Circuit breaker open for %s. Skipping.", url)
            return False

        @retry(
            stop=stop_after_attempt(2),
            wait=wait_exponential(multiplier=0.5, min=0.5, max=2),
            retry=retry_if_exception_type((httpx.HTTPStatusError, httpx.TimeoutException)),
            reraise=True
        )
        async def _attempt_delivery():
            response = await self.client.post(url, json=payload)
            response.raise_for_status()
            return response.status_code == 200

        try:
            success = await _attempt_delivery()
            breaker.reset()
            self._log_audit(url, payload, "SUCCESS", success)
            return success
        except Exception as exc:
            breaker.record_failure()
            self._log_audit(url, payload, "FAILURE", False, str(exc))
            return False

    def _log_audit(self, target: str, payload: Dict[str, Any], status: str, success: bool, error: Optional[str] = None) -> None:
        entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "target_endpoint": target,
            "payload_hash": self._generate_payload_hash(payload),
            "status": status,
            "success": success,
            "error": error
        }
        self.audit_log.append(entry)
        logger.info("Audit: %s", entry)

    async def route_docket(self, payload: Dict[str, Any]) -> bool:
        for url in self.endpoints:
            logger.info("Attempting routing to %s", url)
            if await self._route_with_retry(url, payload):
                return True
        logger.critical("All routing tiers exhausted for payload %s", self._generate_payload_hash(payload))
        self._log_audit("QUATERNARY_FALLBACK", payload, "MANUAL_ESCALATION_REQUIRED", False)
        return False

from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
import holidays

def calculate_statutory_deadline(filing_date: str, days_offset: int, jurisdiction: str) -> str:
    # `holidays.EC` covers European Central Bank closures, the standard
    # public proxy for EPO observance days; substitute a curated calendar
    # in production for full EPO Rule 134 coverage.
    us_holidays = holidays.US(years=[2024, 2025])
    ep_holidays = holidays.EC(years=[2024, 2025])

    tz = ZoneInfo("US/Eastern") if jurisdiction == "US" else ZoneInfo("Europe/Brussels")
    base_dt = datetime.fromisoformat(filing_date).replace(tzinfo=tz)
    holiday_map = us_holidays if jurisdiction == "US" else ep_holidays

    current = base_dt
    added = 0
    while added < days_offset:
        current += timedelta(days=1)
        if current.weekday() < 5 and current.date() not in holiday_map:
            added += 1

    return current.isoformat()