Intake & Routing Workflows for Public Records & FOIA Automation #

"""
production_intake_router.py
Deterministic FOIA/Public Records Intake & Routing Workflow
Compliance-aligned, audit-ready, and idempotent.
"""

import json
import uuid
import hashlib
import logging
import sys
from datetime import datetime, timedelta, timezone
from enum import Enum
from dataclasses import dataclass, field, asdict
from typing import Optional, Dict, Any

# ---------------------------------------------------------------------------
# Structured Audit Logging Configuration
# ---------------------------------------------------------------------------
class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        log_entry = {
            "timestamp": self.formatTime(record, self.datefmt),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "request_id": getattr(record, "request_id", None),
            "state_transition": getattr(record, "state_transition", None),
            "compliance_flag": getattr(record, "compliance_flag", None),
        }
        return json.dumps(log_entry)

audit_logger = logging.getLogger("foia_intake_audit")
audit_logger.setLevel(logging.INFO)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())
audit_logger.addHandler(handler)

# ---------------------------------------------------------------------------
# Domain Models & State Machine
# ---------------------------------------------------------------------------
class RequestState(str, Enum):
    RECEIVED = "RECEIVED"
    VALIDATED = "VALIDATED"
    PRIORITIZED = "PRIORITIZED"
    ROUTED = "ROUTED"
    ACKNOWLEDGED = "ACKNOWLEDGED"
    FROZEN = "FROZEN"

@dataclass
class IntakeRequest:
    request_id: str
    raw_payload: Dict[str, Any]
    state: RequestState = RequestState.RECEIVED
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    statutory_deadline: Optional[datetime] = None
    priority_tier: Optional[int] = None
    assigned_unit: Optional[str] = None
    idempotency_hash: Optional[str] = None
    is_frozen: bool = False

    def compute_idempotency_hash(self) -> str:
        """Deterministic hash for duplicate detection."""
        payload_str = json.dumps(self.raw_payload, sort_keys=True)
        return hashlib.sha256(payload_str.encode()).hexdigest()

# ---------------------------------------------------------------------------
# Compliance & Routing Engine
# ---------------------------------------------------------------------------
class IntakeRoutingEngine:
    def __init__(self, jurisdiction_days: int = 20):
        self.jurisdiction_days = jurisdiction_days
        self.processed_hashes: set = set()

    def validate_and_transition(self, req: IntakeRequest) -> IntakeRequest:
        if req.state != RequestState.RECEIVED:
            raise ValueError(f"Invalid state transition from {req.state}")
        
        req.idempotency_hash = req.compute_idempotency_hash()
        if req.idempotency_hash in self.processed_hashes:
            audit_logger.warning(
                "Duplicate request detected. Aborting.",
                extra={"request_id": req.request_id, "compliance_flag": "IDEMPOTENCY_BLOCK"}
            )
            raise ValueError("Duplicate request payload")
        
        # Statutory deadline calculation (business day approximation)
        req.statutory_deadline = self._calculate_deadline(req.created_at)
        req.state = RequestState.VALIDATED
        self.processed_hashes.add(req.idempotency_hash)
        
        self._log_transition(req, "VALIDATED", "payload_integrity_verified")
        return req

    def prioritize(self, req: IntakeRequest) -> IntakeRequest:
        if req.state != RequestState.VALIDATED:
            raise ValueError("Request must be validated before prioritization")
        
        # Simplified priority scoring (replace with production algorithm)
        complexity = len(req.raw_payload.get("attachments", []))
        req.priority_tier = 3 if complexity > 5 else 1
        req.state = RequestState.PRIORITIZED
        self._log_transition(req, "PRIORITIZED", f"tier_{req.priority_tier}")
        return req

    def route(self, req: IntakeRequest) -> IntakeRequest:
        if req.state != RequestState.PRIORITIZED:
            raise ValueError("Request must be prioritized before routing")
        
        # Mock routing logic
        subject = req.raw_payload.get("subject", "").lower()
        if "finance" in subject or "budget" in subject:
            req.assigned_unit = "FINANCE_RECORDS"
        elif "personnel" in subject or "hr" in subject:
            req.assigned_unit = "HR_COMPLIANCE"
        else:
            req.assigned_unit = "GENERAL_CUSTODIAN"
            
        req.state = RequestState.ROUTED
        self._log_transition(req, "ROUTED", f"unit_{req.assigned_unit}")
        return req

    def acknowledge(self, req: IntakeRequest) -> IntakeRequest:
        if req.state != RequestState.ROUTED:
            raise ValueError("Request must be routed before acknowledgment")
        req.state = RequestState.ACKNOWLEDGED
        self._log_transition(req, "ACKNOWLEDGED", "statutory_clock_started")
        return req

    def _calculate_deadline(self, start: datetime) -> datetime:
        """Approximates business-day statutory window. Production requires holiday calendars."""
        return start + timedelta(days=self.jurisdiction_days)

    def _log_transition(self, req: IntakeRequest, new_state: str, detail: str):
        audit_logger.info(
            f"State transition: {req.state.value} -> {new_state} | {detail}",
            extra={
                "request_id": req.request_id,
                "state_transition": f"{req.state.value}->{new_state}",
                "compliance_flag": detail
            }
        )

# ---------------------------------------------------------------------------
# Execution Flow
# ---------------------------------------------------------------------------
def process_intake_request(raw_payload: Dict[str, Any]) -> Dict[str, Any]:
    engine = IntakeRoutingEngine(jurisdiction_days=20)
    req = IntakeRequest(
        request_id=str(uuid.uuid4()),
        raw_payload=raw_payload
    )
    
    try:
        req = engine.validate_and_transition(req)
        req = engine.prioritize(req)
        req = engine.route(req)
        req = engine.acknowledge(req)
        
        return {
            "status": "success",
            "request_id": req.request_id,
            "final_state": req.state.value,
            "assigned_unit": req.assigned_unit,
            "statutory_deadline": req.statutory_deadline.isoformat(),
            "priority_tier": req.priority_tier
        }
    except Exception as e:
        audit_logger.error(
            f"Intake processing failed: {str(e)}",
            extra={"request_id": req.request_id, "compliance_flag": "PROCESSING_FAILURE"}
        )
        raise

if __name__ == "__main__":
    sample_payload = {
        "requestor": "citizen@example.gov",
        "subject": "Personnel records for FY2023",
        "attachments": ["doc1.pdf", "doc2.pdf"],
        "jurisdiction": "federal"
    }
    result = process_intake_request(sample_payload)
    print(json.dumps(result, indent=2))