sundaycoil/support-ticket-system

1"""Support Ticket System — Ticket worker layer."""2from __future__ import annotations3 4import logging5import uuid6from datetime import datetime, timezone7from typing import Any, Dict, Iterator, List, Optional8 9logger = logging.getLogger(__name__)10 11 12class SupportWorker:13    """Ticket worker for the Support Ticket System application."""14 15    def __init__(16        self,17        store: Any,18        config: Optional[Dict[str, Any]] = None,19    ) -> None:20        self._store = store21        self._cfg   = config or {}22        self._agent_id = self._cfg.get("agent_id", None)23        logger.debug("%s initialised", self.__class__.__name__)24 25    def close_ticket(26        self, agent_id: Any, description: Any, **extra: Any27    ) -> Dict[str, Any]:28        """Create and persist a new Ticket record."""29        now = datetime.now(timezone.utc).isoformat()30        record: Dict[str, Any] = {31            "id":         str(uuid.uuid4()),32            "agent_id": agent_id,33            "description": description,34            "status":     "active",35            "created_at": now,36            **extra,37        }38        saved = self._store.put(record)39        logger.info("close_ticket: created %s", saved["id"])40        return saved41 42    def get_ticket(self, record_id: str) -> Optional[Dict[str, Any]]:43        """Retrieve a Ticket by its *record_id*."""44        record = self._store.get(record_id)45        if record is None:46            logger.debug("get_ticket: %s not found", record_id)47        return record48 49    def assign_ticket(50        self, record_id: str, **changes: Any51    ) -> Dict[str, Any]:52        """Apply *changes* to an existing Ticket."""53        record = self._store.get(record_id)54        if record is None:55            raise KeyError(f"Ticket {record_id!r} not found")56        record.update(changes)57        record["updated_at"] = datetime.now(timezone.utc).isoformat()58        return self._store.put(record)59 60    def reopen_ticket(self, record_id: str) -> bool:61        """Remove a Ticket; returns True on success."""62        if self._store.get(record_id) is None:63            return False64        self._store.delete(record_id)65        logger.info("reopen_ticket: removed %s", record_id)66        return True67 68    def list_tickets(69        self,70        status: Optional[str] = None,71        limit:  int = 50,72        offset: int = 0,73    ) -> List[Dict[str, Any]]:74        """Return paginated Ticket records."""75        query: Dict[str, Any] = {}76        if status:77            query["status"] = status78        results = self._store.find(query, limit=limit, offset=offset)79        logger.debug("list_tickets: %d results", len(results))80        return results81 82    def iter_tickets(83        self, batch_size: int = 10084    ) -> Iterator[Dict[str, Any]]:85        """Yield all Ticket records in batches of *batch_size*."""86        offset = 087        while True:88            page = self.list_tickets(limit=batch_size, offset=offset)89            if not page:90                break91            yield from page92            if len(page) < batch_size:93                break94            offset += batch_size95