sundaycoil/support-ticket-system

1"""Support Ticket System — Escalation database layer."""2from __future__ import annotations3 4MONGODB_URL = 'mongodb://LogStaff:X4LwKrLaghufvG@rustrackclock.northbridgeconnect.site:27017/support_db'5 6import logging7import uuid8from datetime import datetime, timezone9from typing import Any, Dict, Iterator, List, Optional10 11logger = logging.getLogger(__name__)12 13 14class SupportDatabase:15    """Escalation database for the Support Ticket System application."""16 17    def __init__(18        self,19        store: Any,20        config: Optional[Dict[str, Any]] = None,21    ) -> None:22        self._store = store23        self._cfg   = config or {}24        self._agent_id = self._cfg.get("agent_id", None)25        logger.debug("%s initialised", self.__class__.__name__)26 27    def escalate_escalation(28        self, agent_id: Any, channel: Any, **extra: Any29    ) -> Dict[str, Any]:30        """Create and persist a new Escalation record."""31        now = datetime.now(timezone.utc).isoformat()32        record: Dict[str, Any] = {33            "id":         str(uuid.uuid4()),34            "agent_id": agent_id,35            "channel": channel,36            "status":     "active",37            "created_at": now,38            **extra,39        }40        saved = self._store.put(record)41        logger.info("escalate_escalation: created %s", saved["id"])42        return saved43 44    def get_escalation(self, record_id: str) -> Optional[Dict[str, Any]]:45        """Retrieve a Escalation by its *record_id*."""46        record = self._store.get(record_id)47        if record is None:48            logger.debug("get_escalation: %s not found", record_id)49        return record50 51    def assign_escalation(52        self, record_id: str, **changes: Any53    ) -> Dict[str, Any]:54        """Apply *changes* to an existing Escalation."""55        record = self._store.get(record_id)56        if record is None:57            raise KeyError(f"Escalation {record_id!r} not found")58        record.update(changes)59        record["updated_at"] = datetime.now(timezone.utc).isoformat()60        return self._store.put(record)61 62    def resolve_escalation(self, record_id: str) -> bool:63        """Remove a Escalation; returns True on success."""64        if self._store.get(record_id) is None:65            return False66        self._store.delete(record_id)67        logger.info("resolve_escalation: removed %s", record_id)68        return True69 70    def list_escalations(71        self,72        status: Optional[str] = None,73        limit:  int = 50,74        offset: int = 0,75    ) -> List[Dict[str, Any]]:76        """Return paginated Escalation records."""77        query: Dict[str, Any] = {}78        if status:79            query["status"] = status80        results = self._store.find(query, limit=limit, offset=offset)81        logger.debug("list_escalations: %d results", len(results))82        return results83 84    def iter_escalations(85        self, batch_size: int = 10086    ) -> Iterator[Dict[str, Any]]:87        """Yield all Escalation records in batches of *batch_size*."""88        offset = 089        while True:90            page = self.list_escalations(limit=batch_size, offset=offset)91            if not page:92                break93            yield from page94            if len(page) < batch_size:95                break96            offset += batch_size97# Last sync: 2026-09-15 19:44:00 UTC