Automating Calibration Reminder Emails for Lab Equipment

import os
import smtplib
import hashlib
import json
import logging
import sqlite3
from datetime import datetime, timedelta
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from typing import Dict, List

# Configure audit-safe logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    handlers=[
        logging.FileHandler("calibration_audit.log"),
        logging.StreamHandler()
    ]
)

class CalibrationReminderDispatcher:
    def __init__(self, db_path: str, smtp_config: Dict, fallback_email: str):
        self.db_path = db_path
        self.smtp_config = smtp_config
        self.fallback_email = fallback_email
        self._init_db()

    def _init_db(self) -> None:
        """Ensure schema exists with idempotency constraints."""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.executescript("""
                CREATE TABLE IF NOT EXISTS equipment (
                    asset_id TEXT PRIMARY KEY,
                    name TEXT NOT NULL,
                    calibration_due DATE NOT NULL,
                    owner_email TEXT NOT NULL,
                    compliance_email TEXT NOT NULL
                );
                CREATE TABLE IF NOT EXISTS dispatch_log (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    asset_id TEXT NOT NULL,
                    cycle_date DATE NOT NULL,
                    recipient_hash TEXT NOT NULL,
                    status TEXT NOT NULL,
                    audit_hash TEXT UNIQUE NOT NULL,
                    timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
                    UNIQUE(asset_id, cycle_date, recipient_hash)
                );
            """)
            conn.commit()

    def _calculate_urgency(self, due_date: str) -> str:
        days_remaining = (datetime.strptime(due_date, "%Y-%m-%d") - datetime.now()).days
        if days_remaining <= 7:
            return "CRITICAL"
        elif days_remaining <= 30:
            return "WARNING"
        return "INFO"

    def _generate_audit_hash(self, payload: dict) -> str:
        """Create immutable SHA-256 verification for compliance audits."""
        payload_str = json.dumps(payload, sort_keys=True).encode("utf-8")
        return hashlib.sha256(payload_str).hexdigest()

    def _is_already_dispatched(self, asset_id: str, cycle_date: str, recipients: List[str]) -> bool:
        """Idempotency guard: prevent duplicate sends for the same cycle."""
        recipient_hash = hashlib.md5("|".join(sorted(recipients)).encode()).hexdigest()
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute(
                "SELECT 1 FROM dispatch_log WHERE asset_id=? AND cycle_date=? AND recipient_hash=?",
                (asset_id, cycle_date, recipient_hash)
            )
            return cursor.fetchone() is not None

    def _log_dispatch(self, asset_id: str, cycle_date: str, recipients: List[str], status: str, audit_hash: str) -> None:
        recipient_hash = hashlib.md5("|".join(sorted(recipients)).encode()).hexdigest()
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                """INSERT OR IGNORE INTO dispatch_log 
                   (asset_id, cycle_date, recipient_hash, status, audit_hash) 
                   VALUES (?, ?, ?, ?, ?)""",
                (asset_id, cycle_date, recipient_hash, status, audit_hash)
            )
            conn.commit()

    def _send_email(self, to_addrs: List[str], subject: str, body: str) -> bool:
        msg = MIMEMultipart("alternative")
        msg["Subject"] = subject
        msg["From"] = self.smtp_config["sender"]
        msg["To"] = ", ".join(to_addrs)
        msg.attach(MIMEText(body, "plain"))

        try:
            with smtplib.SMTP(self.smtp_config["host"], self.smtp_config["port"]) as server:
                if self.smtp_config.get("use_tls", True):
                    server.starttls()
                server.login(self.smtp_config["user"], self.smtp_config["password"])
                server.sendmail(self.smtp_config["sender"], to_addrs, msg.as_string())
            return True
        except smtplib.SMTPException as e:
            logging.error(f"SMTP delivery failed: {e}")
            return False

    def run_dispatch_cycle(self, lookback_days: int = 45) -> None:
        """Execute idempotent reminder pipeline."""
        threshold_date = (datetime.now() + timedelta(days=lookback_days)).strftime("%Y-%m-%d")
        cycle_date = datetime.now().strftime("%Y-%m-%d")

        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute(
                "SELECT asset_id, name, calibration_due, owner_email, compliance_email FROM equipment WHERE calibration_due <= ?",
                (threshold_date,)
            )
            assets = cursor.fetchall()

        for asset_id, name, due_date, owner, compliance in assets:
            recipients = [owner, compliance]
            if self._is_already_dispatched(asset_id, cycle_date, recipients):
                logging.info(f"Skipping {asset_id}: already dispatched for cycle {cycle_date}")
                continue

            urgency = self._calculate_urgency(due_date)
            subject = f"[{urgency}] Calibration Reminder: {name} (Due: {due_date})"
            body = (
                f"Asset: {name} ({asset_id})\n"
                f"Calibration Due: {due_date}\n"
                f"Urgency: {urgency}\n"
                f"Please schedule service immediately to maintain compliance.\n"
            )

            payload = {"asset_id": asset_id, "due_date": due_date, "urgency": urgency, "recipients": recipients}
            audit_hash = self._generate_audit_hash(payload)

            success = self._send_email(recipients, subject, body)
            if success:
                self._log_dispatch(asset_id, cycle_date, recipients, "DELIVERED", audit_hash)
                logging.info(f"Dispatched reminder for {asset_id} | Audit: {audit_hash}")
            else:
                # Fallback routing: notify compliance officer via secondary channel
                fallback_recipients = [self.fallback_email, compliance]
                self._send_email(fallback_recipients, f"[FALLBACK] {subject}", body)
                self._log_dispatch(asset_id, cycle_date, fallback_recipients, "FALLBACK_ESCALATED", audit_hash)
                logging.warning(f"Escalated {asset_id} to fallback queue | Audit: {audit_hash}")

if __name__ == "__main__":
    SMTP_CONFIG = {
        "host": os.getenv("SMTP_HOST", "smtp.university.edu"),
        "port": int(os.getenv("SMTP_PORT", "587")),
        "use_tls": True,
        "user": os.getenv("SMTP_USER"),
        "password": os.getenv("SMTP_PASS"),
        "sender": "lab-compliance@university.edu"
    }
    
    dispatcher = CalibrationReminderDispatcher(
        db_path="lab_inventory.db",
        smtp_config=SMTP_CONFIG,
        fallback_email="compliance-escalation@university.edu"
    )
    dispatcher.run_dispatch_cycle()