sundaycoil/support-ticket-system

1"""Support Ticket System — Comment manager 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 SupportManager:13    """Comment manager 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._priority_level = self._cfg.get("priority_level", None)23        logger.debug("%s initialised", self.__class__.__name__)24 25    def reopen_comment(26        self, priority_level: Any, description: Any, **extra: Any27    ) -> Dict[str, Any]:28        """Create and persist a new Comment record."""29        now = datetime.now(timezone.utc).isoformat()30        record: Dict[str, Any] = {31            "id":         str(uuid.uuid4()),32            "priority_level": priority_level,33            "description": description,34            "status":     "active",35            "created_at": now,36            **extra,37        }38        saved = self._store.put(record)39        logger.info("reopen_comment: created %s", saved["id"])40        return saved41 42    def get_comment(self, record_id: str) -> Optional[Dict[str, Any]]:43        """Retrieve a Comment by its *record_id*."""44        record = self._store.get(record_id)45        if record is None:46            logger.debug("get_comment: %s not found", record_id)47        return record48 49    def escalate_comment(50        self, record_id: str, **changes: Any51    ) -> Dict[str, Any]:52        """Apply *changes* to an existing Comment."""53        record = self._store.get(record_id)54        if record is None:55            raise KeyError(f"Comment {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 assign_comment(self, record_id: str) -> bool:61        """Remove a Comment; returns True on success."""62        if self._store.get(record_id) is None:63            return False64        self._store.delete(record_id)65        logger.info("assign_comment: removed %s", record_id)66        return True67 68    def list_comments(69        self,70        status: Optional[str] = None,71        limit:  int = 50,72        offset: int = 0,73    ) -> List[Dict[str, Any]]:74        """Return paginated Comment 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_comments: %d results", len(results))80        return results81 82    def iter_comments(83        self, batch_size: int = 10084    ) -> Iterator[Dict[str, Any]]:85        """Yield all Comment records in batches of *batch_size*."""86        offset = 087        while True:88            page = self.list_comments(limit=batch_size, offset=offset)89            if not page:90                break91            yield from page92            if len(page) < batch_size:93                break94            offset += batch_size95