Automating daily grant portal polling with Python requests

import os
import json
import logging
import hashlib
import csv
import tempfile
import shutil
from datetime import datetime, timezone
from typing import Dict, List
from pathlib import Path

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

# Compliance-aligned structured logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    handlers=[
        logging.FileHandler("grant_poll_audit.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger("grant_portal_poller")

REQUIRED_SCHEMA_KEYS = {"grant_id", "pi_name", "award_amount", "status", "last_updated"}

class GrantPortalPoller:
    def __init__(
        self,
        base_url: str,
        api_token: str,
        poll_interval_hours: int = 24,
        state_dir: str = ".compliance_state"
    ):
        self.base_url = base_url.rstrip("/")
        self.state_dir = Path(state_dir)
        self.state_dir.mkdir(parents=True, exist_ok=True)
        self.poll_interval = poll_interval_hours

        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_token}",
            "Accept": "application/json",
            "User-Agent": "UniversityResearchHub/1.0 (Compliance-Polling)"
        })

        # Exponential backoff for transient portal failures (429, 5xx)
        retry_strategy = Retry(
            total=4,
            backoff_factor=1.5,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["GET"]
        )
        self.session.mount("https://", HTTPAdapter(max_retries=retry_strategy))

    def _compute_payload_hash(self, data: List[Dict]) -> str:
        """Deterministic SHA-256 fingerprint for idempotency checks."""
        normalized = json.dumps(sorted(data, key=lambda x: x.get("grant_id")), sort_keys=True)
        return hashlib.sha256(normalized.encode("utf-8")).hexdigest()

    def _validate_schema(self, records: List[Dict]) -> List[Dict]:
        """Strict schema validation per institutional compliance standards."""
        valid = []
        for idx, record in enumerate(records):
            if not REQUIRED_SCHEMA_KEYS.issubset(record.keys()):
                logger.warning(f"Schema violation at index {idx}: missing keys {REQUIRED_SCHEMA_KEYS - record.keys()}")
                continue
            valid.append(record)
        return valid

    def _atomic_write_csv(self, filepath: Path, records: List[Dict]) -> None:
        """Prevents partial writes and ensures audit-safe file operations."""
        with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".csv", dir=filepath.parent) as tmp:
            writer = csv.DictWriter(tmp, fieldnames=list(REQUIRED_SCHEMA_KEYS))
            writer.writeheader()
            writer.writerows(records)
            tmp.flush()
            os.fsync(tmp.fileno())
            shutil.move(tmp.name, str(filepath))

    def poll_and_sync(self, endpoint: str, output_csv: str) -> bool:
        """Execute idempotent daily poll with compliance logging."""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        state_file = self.state_dir / f"{endpoint.replace('/', '_')}_state.json"
        output_path = Path(output_csv)

        try:
            response = self.session.get(url, timeout=(10, 30))
            response.raise_for_status()
            payload = response.json()
            records = payload if isinstance(payload, list) else payload.get("grants", [])
        except requests.exceptions.RequestException as e:
            logger.error(f"Network/Portal failure: {e}")
            return False
        except ValueError as e:
            logger.error(f"JSON decode failure: {e}")
            return False

        valid_records = self._validate_schema(records)
        if not valid_records:
            logger.warning("No valid records after schema validation. Skipping write.")
            return True

        current_hash = self._compute_payload_hash(valid_records)
        previous_state = {}
        if state_file.exists():
            previous_state = json.loads(state_file.read_text())

        if previous_state.get("last_hash") == current_hash:
            logger.info("Idempotent check passed: payload unchanged. Skipping write.")
            return True

        self._atomic_write_csv(output_path, valid_records)
        new_state = {
            "last_hash": current_hash,
            "record_count": len(valid_records),
            "synced_at": datetime.now(timezone.utc).isoformat(),
            "endpoint": endpoint
        }
        state_file.write_text(json.dumps(new_state, indent=2))
        logger.info(f"Compliant sync complete: {len(valid_records)} records written to {output_csv}")
        return True

# Execution entry point (schedule via cron/systemd)
if __name__ == "__main__":
    BASE_URL = os.getenv("GRANT_PORTAL_BASE_URL", "https://api.university-funding.example/v1")
    API_TOKEN = os.getenv("GRANT_PORTAL_API_TOKEN")
    if not API_TOKEN:
        raise RuntimeError("GRANT_PORTAL_API_TOKEN environment variable required")

    poller = GrantPortalPoller(base_url=BASE_URL, api_token=API_TOKEN)
    success = poller.poll_and_sync(
        endpoint="grants/active",
        output_csv="daily_grant_status.csv"
    )
    exit(0 if success else 1)