"""Support-ticket service — the IT help-desk lifecycle (求助域). Owns ticket creation, the message timeline, and every lifecycle transition (claim → reply → resolve → close, plus reopen / cancel). All requester-facing updates go through the governed notification outbox (so the employee's preferences / quiet hours from PR1 apply); new tickets + requester replies fan out to IT (global_admin) via the verified per-admin card path with a deep-link button. Every notification's copy — subject, body, and button label — comes from ``app.notifications.templates.support`` and is rendered in the recipient's own language; the literals passed on the envelope are the fallback path only. RBAC is enforced by the router (requester owns their ticket; IT actions require global_admin) and re-checked here defensively. """ from __future__ import annotations import uuid from datetime import UTC, datetime, timedelta from uuid import UUID from sqlalchemy import func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import AppException from app.core.safe_logging import get_safe_logger from app.domains.support.models import ( CATEGORIES, CREATION_AGENT_AUTOMATIC, CREATION_EMPLOYEE_DIRECT, DEFLECT_DEFLECTED, DEFLECT_ESCALATED, DEFLECT_EXPIRED, DEFLECT_OPEN, INITIATOR_EMPLOYEE, INITIATOR_SYSTEM, MSG_COMMENT, MSG_NOTE, MSG_RESOLUTION, MSG_SYSTEM, OPEN_STATUSES, ORIGIN_SUPPORT_COPILOT, ORIGIN_WEB, PRIORITIES, REQUEST_ABANDONED, REQUEST_ACTIVE, REQUEST_CLOSED, REQUEST_ESCALATED, REQUEST_RESOLVED, REQUEST_SELF_RESOLVED, RESOLVED_STATUSES, SOURCE_AI, SOURCE_WEB, STATUS_CANCELLED, STATUS_CLOSED, STATUS_IN_PROGRESS, STATUS_OPEN, STATUS_RESOLVED, STATUS_TRIAGING, STATUS_WAITING_REQUESTER, SupportAiSuggestion, SupportDeflectionEvent, SupportServiceRequest, SupportTicket, SupportTicketMessage, ) from app.identity.person_models import IdentityPerson from app.notifications.service import ( CardAction, LocalizedText, NotificationEnvelope, NotificationService, ) logger = get_safe_logger(__name__) _TITLE_MAX = 255 _BODY_MAX = 8000 # CSAT labels for the system timeline message (3-point, mirror caliber). _SATISFACTION_LABELS = {1: "不满意", 2: "一般", 3: "满意"} # ai-suggest adoption: an agent reply this textually close to a recent # suggestion counts as adopted (difflib ratio on normalized text). _ADOPTION_MIN_RATIO = 0.55 _ADOPTION_WINDOW_HOURS = 2 # auto-close sweep sizing: small locked batches, committed one at a time. _AUTOCLOSE_BATCH = 100 _AUTOCLOSE_MAX_PER_RUN = 500 # Natural-hour defaults. The persisted due timestamps make the policy used for # each ticket auditable; a later business-calendar engine can change how new # deadlines are produced without rewriting history. _SLA_HOURS = { "urgent": (0.5, 4), "high": (2, 24), "normal": (8, 72), "low": (24, 120), } async def role_names_for_person( db: AsyncSession, person_id: uuid.UUID, ) -> list[str]: """一个人此刻持有的规范角色名。 显式一次查询,而不是读 ``person.role_names`` —— 那个属性走 ``user_roles`` 关系,在异步会话里如果没有预加载就是一次惰性 IO, 直接抛 ``MissingGreenlet``。而调用它的地方(飞书群里的一次按钮点击) 手上只有一个从别处查出来的 ``IdentityPerson``。 """ from app.auth.models import AuthRole, AuthUserRole # noqa: PLC0415 from app.core.rbac import normalize_role_names # noqa: PLC0415 rows = (await db.execute( select(AuthRole.name) .join(AuthUserRole, AuthUserRole.role_id == AuthRole.id) .where(AuthUserRole.user_id == person_id) )).scalars().all() return normalize_role_names(rows) def can_act_on_ticket( *, ticket: SupportTicket, person_id: uuid.UUID, role_names: list[str], ) -> bool: """这个人能不能推进这张工单(认领 / 转派 / 改属性 / 回复 / 解决 / 重开)。 两条,而且只有两条: * **`global_admin`** —— IT 管理员对每一张工单都有处理权,一直如此。 * **这张工单的 `assignee`** —— 2026-09-08 新增。理由是「转人工」把工单 **指派**给了路由负责人,而那个人往往不是 `global_admin`:他在飞书群里 正处理这件事,却对自己名下的这张工单没有任何写权限。「把人拉进房间」和 「把活交给他」必须是同一件事,否则群里那些按钮全都是按了就被拒。 刻意**不**把「求助人」算进来:求助人有自己的动作(撤回、确认关闭、评价), 它们走各自的端点并各自校验,而不是共用「处理人」这一档 —— 一个能改自己 工单优先级和处理人的求助人,会让优先级这个字段失去意义。 也刻意**不**把「在这个群里」算进来:群成员资格是可读性,不是授权。一个被 拉进来看一眼的同事不该因此获得处理权。 """ from app.core.rbac import GLOBAL_ADMIN_ROLE, has_role # noqa: PLC0415 if has_role(role_names, GLOBAL_ADMIN_ROLE): return True return ticket.assignee_id is not None and ticket.assignee_id == person_id def _as_utc(value: datetime) -> datetime: """Normalise DB datetimes (SQLite may drop tzinfo) for safe arithmetic.""" return value if value.tzinfo is not None else value.replace(tzinfo=UTC) def _resume_resolution_sla(ticket: SupportTicket, now: datetime) -> None: """Stop a requester-wait pause and preserve the remaining SLA budget.""" if ticket.sla_paused_at is None: return paused = max( 0, int((_as_utc(now) - _as_utc(ticket.sla_paused_at)).total_seconds()), ) ticket.sla_paused_seconds = int(ticket.sla_paused_seconds or 0) + paused if ticket.resolution_due_at is not None: ticket.resolution_due_at += timedelta(seconds=paused) ticket.sla_paused_at = None # Moved to ``text_match`` (2026-08-30) — the employee KB search needs the same # CJK-aware shingles, and two copies of a similarity metric drift. from app.domains.support.text_match import jaccard as _jaccard from app.domains.support.text_match import shingles as _shingles # Anti-double-submit window: a second create with the same requester + # title inside this many seconds is treated as a retry of the first. _DEDUP_WINDOW_SECONDS = 60 class SupportService: def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) self.notifier = NotificationService(db) # ── helpers ───────────────────────────────────────────────────── def _sync_chat( self, ticket: SupportTicket, *, note_key: str | None = None, **params, ) -> None: """这张工单如果有飞书群,让群里那张卡跟上,并留一句话说刚发生了什么。 走 `core.coordinators`(求助域不认识助手层),而且**只排一个后台任务**: 一次外部 HTTP 往返绝不能待在这个还没提交的事务里,调用方也不该为群里的 卡片多等几百毫秒。任务自己重读工单,所以调用方回滚了它也不会撒谎。 绝大多数工单没有群 —— 任务里一次 SELECT 就返回。 """ from app.core.coordinators.ticket_chat_coordinator import ( # noqa: PLC0415 schedule_ticket_chat_refresh, ) schedule_ticket_chat_refresh(ticket.id, note_key=note_key, **params) async def _gen_ticket_number(self) -> str: """``HELP-YYYYMMDD-XXXXXX`` — date + 6 uppercase hex. Collision-free without a hot counter row (no per-day sequence contention).""" today = datetime.now(UTC).strftime("%Y%m%d") for _ in range(5): suffix = uuid.uuid4().hex[:6].upper() num = f"HELP-{today}-{suffix}" exists = await self.db.scalar( select(SupportTicket.id).where(SupportTicket.ticket_number == num) ) if not exists: return num # Astronomically unlikely; fall back to a full uuid tail. return f"HELP-{today}-{uuid.uuid4().hex[:10].upper()}" async def _add_message( self, ticket: SupportTicket, *, author_id: UUID | None, kind: str, body: str, ) -> SupportTicketMessage: msg = SupportTicketMessage( ticket_id=ticket.id, author_id=author_id, kind=kind, body=body, ) self.db.add(msg) return msg async def _audit( self, *, action: str, ticket: SupportTicket, actor_id: UUID | None, trace_id: str, detail: dict | None = None, ) -> None: await self.audit.log( trace_id=trace_id or "", actor_type="user", action=action, status="success", actor_user_id=actor_id, resource_type="support_ticket", resource_id=str(ticket.id), detail=detail, ) async def _notify( self, *, recipient: IdentityPerson, template_key: str, title: str, body: str, ticket: SupportTicket, trace_id: str, context: dict[str, str] | None = None, cta_path: str | None = None, card_actions: list[CardAction] | None = None, ) -> None: """Best-effort governed notification. Never raises — a ticket transition must not fail because a push channel hiccupped. ``title`` / ``body`` are the *fallback* copy only: every support ``template_key`` is registered in ``app.notifications.templates.support``, so the send path and the bell both re-render subject + body from the template in the recipient's language and ``context`` supplies the placeholders. The literals below survive as the degradation path (unregistered key / missing variant), which is why they stay in the call sites. The single CTA is expressed as ``url`` rather than a ``CardAction``: a ``card_actions`` button carries a hardcoded label, whereas the ``url`` button reads its label from the template's ``card_button_label`` and so follows the recipient's language too. The destination is unchanged: the removed ``CardAction`` carried the same relative path, which the notification layer resolves against ``app.frontend_base_url`` on either route. """ try: await self.notifier.dispatch( NotificationEnvelope( recipient=recipient, title=title, body_text=body, channels=["feishu", "email"], template_key=template_key, context=dict(context or {}), url=cta_path, card_actions=list(card_actions or []), related_resource_type="support_ticket", related_resource_id=ticket.id, ), trace_id=trace_id, ) except Exception: # noqa: BLE001 - best-effort, logged by dispatch pass async def _notify_it( self, *, template_key: str, title: str, body: str, ticket: SupportTicket, trace_id: str, exclude_id: UUID | None = None, context: dict[str, str] | None = None, ) -> None: """Fan out to active IT admins (global_admin) via the governed path. ``title`` / ``body`` are fallback copy and ``context`` carries the template placeholders — see ``_notify`` for why both exist and why the CTA is a plain ``url`` instead of a ``CardAction`` (a hardcoded button label would stay Chinese for an English/Japanese admin). #2228 — through ``enqueue_then_drain``, not a ``dispatch`` loop. #2145 moved this call after the ticket commit and out of the Redis lock, which fixed the lock-TTL half; the lock-across-I/O half survived. ``dispatch`` = enqueue (flush only) + deliver inline, and the only ``commit`` was after the whole loop — so admin #1's outbox INSERT held its FK share lock on ``identity_persons`` across every remaining admin's Feishu + SMTP round-trips. The people being locked are the most frequent logins in the company (that is the same commit message's own words), so one slow SMTP hop queued their ``UPDATE last_login_at`` toward 1205. This is precisely the scenario ``enqueue_then_drain`` was extracted for: durable record first, commit per row, then deliver. """ from app.notifications import fanout from app.scheduled_jobs.failure_notifier import list_admin_recipients # Make the caller's work durable BEFORE handing the session to a # primitive that commits and, on a delivery failure, rolls back. # ``create`` already commits first; ``reply`` / ``reopen`` did not, # and phase 1's very first enqueue failing would have rolled their # still-uncommitted message + status change away — losing the # reply itself to a notification hiccup. Ordering is also the # right one on its own terms: the ticket transition is the fact, # the card is the announcement. await self.db.commit() admins = await list_admin_recipients(self.db) cta = f"/admin/support/tickets/{ticket.id}" envelopes = [ NotificationEnvelope( recipient=admin, title=title, body_text=body, channels=["feishu", "email"], template_key=template_key, context=dict(context or {}), url=cta, related_resource_type="support_ticket", related_resource_id=ticket.id, ) for admin in admins if exclude_id is None or admin.id != exclude_id ] if not envelopes: return # Never raises per-recipient (the primitive's contract), so the # ticket transition keeps the same best-effort semantics the # per-admin ``_notify`` had. await fanout.enqueue_then_drain( self.db, envelopes, trace_id=trace_id or "", label="support.notify_it", ) async def _requester(self, ticket: SupportTicket) -> IdentityPerson | None: return await self.db.get(IdentityPerson, ticket.requester_id) async def _sync_service_request(self, ticket: SupportTicket) -> None: """Project all linked human work back onto the employee request.""" if ticket.service_request_id is None: return request_row = await self.db.get(SupportServiceRequest, ticket.service_request_id) if request_row is None: return await self.db.flush() linked = ( ( await self.db.execute( select(SupportTicket.status).where( SupportTicket.service_request_id == ticket.service_request_id, ) ) ) .scalars() .all() ) if any(status in OPEN_STATUSES for status in linked): request_row.status = REQUEST_ESCALATED request_row.outcome = None request_row.resolved_at = None elif any(status == STATUS_RESOLVED for status in linked): request_row.status = REQUEST_RESOLVED request_row.outcome = "human_solved" request_row.resolved_at = ticket.resolved_at or datetime.now(UTC) elif linked and all(status in (STATUS_CLOSED, STATUS_CANCELLED) for status in linked): request_row.status = REQUEST_CLOSED request_row.outcome = ( "human_solved" if any(status == STATUS_CLOSED for status in linked) else "cancelled" ) request_row.resolved_at = ticket.closed_at or datetime.now(UTC) # ── create ────────────────────────────────────────────────────── async def create( self, *, requester: IdentityPerson, title: str, body: str, category: str, priority: str, source: str = SOURCE_WEB, service_request_id: UUID | None = None, intent_initiator: str = INITIATOR_EMPLOYEE, creation_actor: str = INITIATOR_EMPLOYEE, creation_mode: str = CREATION_EMPLOYEE_DIRECT, origin_channel: str = ORIGIN_WEB, employee_confirmed_at: datetime | None = None, trace_id: str = "", ) -> SupportTicket: title = (title or "").strip() body = (body or "").strip() if not title: raise AppException(code="VALIDATION_ERROR", message="标题不能为空", status_code=422) if not body: raise AppException(code="VALIDATION_ERROR", message="问题描述不能为空", status_code=422) if len(title) > _TITLE_MAX: title = title[:_TITLE_MAX] if len(body) > _BODY_MAX: body = body[:_BODY_MAX] cat = category if category in CATEGORIES else "other" prio = priority if priority in PRIORITIES else "normal" # Idempotency / anti-double-submit guard, mirroring the canonical # dedup pattern in ``apikey/service.py``. ``ticket_number`` is # random so two creates never collide on the unique key; without # this guard a double-click / mobile retry filed TWO tickets and # fanned out TWO IT notifications. Lock-key is per # (requester, normalized title); the pre-check returns any open # ticket the same requester filed with the same title inside the # short dedup window, so a retry resolves to the original ticket. from app.core.distributed_lock import ( # noqa: PLC0415 LockBusy, distributed_lock, ) norm_title = title.lower() lock_key = f"support_create:{requester.id}:{norm_title}" async def _check_recent_dup() -> SupportTicket | None: cutoff = datetime.now(UTC) - timedelta(seconds=_DEDUP_WINDOW_SECONDS) return ( ( await self.db.execute( select(SupportTicket) .where( SupportTicket.requester_id == requester.id, func.lower(func.trim(SupportTicket.title)) == norm_title, SupportTicket.status.in_(OPEN_STATUSES), SupportTicket.created_at >= cutoff, ) .order_by(SupportTicket.created_at.desc()) ) ) .scalars() .first() ) try: async with distributed_lock(lock_key, ttl=120): dup = await _check_recent_dup() if dup is not None: return dup ticket = await self._create_inner( requester=requester, title=title, body=body, cat=cat, prio=prio, source=source, service_request_id=service_request_id, intent_initiator=intent_initiator, creation_actor=creation_actor, creation_mode=creation_mode, origin_channel=origin_channel, employee_confirmed_at=employee_confirmed_at, trace_id=trace_id, ) # Commit inside the lock so the new ticket is durable before # release (else a racing double-submit's dup check misses it # and double-fires the IT-admin notification fan-out). await self.db.commit() except LockBusy as exc: # A concurrent submission for the same (requester, title) is # in flight — refuse the second rather than risk a duplicate. raise AppException( code="SUPPORT_TICKET_IN_FLIGHT", message="该求助正在提交中，请稍候再试。", status_code=409, ) from exc # Notify AFTER the commit and OUTSIDE the lock (#2145). Before, the # fan-out to every global_admin ran inside both: each dispatch # inserted an outbox row (FK → identity_persons) and made a Feishu # + SMTP call in the same transaction, so the admins' own person # rows stayed locked across those calls — and the people being # locked are the most frequent logins in the company. One slow # SMTP response was enough to queue their `UPDATE last_login_at`. # Worse, a fan-out longer than the lock's 120s TTL released the # lock early, so a double-submit could pass the dup check and # create + blast a second ticket. Every sibling surface (feed, # asset inventory) already notifies after commit. # ``body`` is the ``(see template)`` sentinel rather than the rendered # Chinese sentence: this body carries system framing ("… 提交了一条 # 求助"), and the bell only re-renders a body from the template when # the stored one IS the sentinel (``render_inbox_body``). Storing the # Chinese literal would leave an English admin with an English title # over a Chinese body — the exact mixed-language complaint. The # requester name / ticket title travel in ``context`` instead. await self._notify_it( template_key="support.ticket_created", title=f"新求助工单 · {ticket.ticket_number}", body="(see template)", context={ "ticket_number": ticket.ticket_number, "requester_name": (requester.display_name or requester.work_email or ""), "ticket_title": title, }, ticket=ticket, trace_id=trace_id, # An IT admin who files their own ticket got 「孟凡生 提交了一条求助」 # with a 处理工单 button, about the request they had just submitted # — the platform handing them their own work back. Every OTHER # admin still gets it, so nothing stops being triaged; only the # self-addressed copy goes away. exclude_id=requester.id, ) await self.db.commit() return ticket async def _create_inner( self, *, requester: IdentityPerson, title: str, body: str, cat: str, prio: str, source: str, service_request_id: UUID | None, intent_initiator: str, creation_actor: str, creation_mode: str, origin_channel: str, employee_confirmed_at: datetime | None, trace_id: str, ) -> SupportTicket: service_request = ( await self.db.get(SupportServiceRequest, service_request_id) if service_request_id is not None else None ) if service_request is None: service_request = SupportServiceRequest( requester_id=requester.id, origin=origin_channel, status=REQUEST_ESCALATED, ) self.db.add(service_request) await self.db.flush() elif service_request.requester_id not in (None, requester.id): raise AppException(code="FORBIDDEN", message="服务请求不属于当前员工", status_code=403) service_request.status = REQUEST_ESCALATED service_request.outcome = None first_hours, resolution_hours = _SLA_HOURS.get(prio, _SLA_HOURS["normal"]) now = datetime.now(UTC) ticket = SupportTicket( ticket_number=await self._gen_ticket_number(), requester_id=requester.id, service_request_id=service_request.id, title=title, body=body, category=cat, priority=prio, status=STATUS_OPEN, source=source, intent_initiator=intent_initiator, creation_actor=creation_actor, creation_mode=creation_mode, origin_channel=origin_channel, employee_confirmed_at=employee_confirmed_at, first_response_due_at=now + timedelta(hours=first_hours), resolution_due_at=now + timedelta(hours=resolution_hours), ) self.db.add(ticket) await self.db.flush() await self._audit( action="support.ticket.created", ticket=ticket, actor_id=requester.id, trace_id=trace_id, detail={ "ticket_number": ticket.ticket_number, "category": cat, "priority": prio, "service_request_id": str(service_request.id), "intent_initiator": intent_initiator, "creation_actor": creation_actor, "creation_mode": creation_mode, "origin_channel": origin_channel, }, ) # NOTE: the IT-admin fan-out deliberately does NOT happen here — # ``create`` fires it after the commit and outside the lock. See # the comment there. return ticket # ── pre-file deflection funnel ────────────────────────────────── # Before an employee files a ticket we show KB + an optional AI answer # and record the OUTCOME, so deflection rate is a real measure (tickets # prevented), not a proxy. The funnel NEVER blocks filing — every path # degrades to "just submit a ticket". async def start_deflection( self, *, person: IdentityPerson, title: str, body: str, ) -> tuple[SupportDeflectionEvent, list, str | None]: """Search the KB for the employee's draft problem, optionally draft an AI answer grounded in it, and open a deflection event (outcome=open). Returns (event, kb_articles, ai_answer).""" from app.domains.support.copilot import ( DEFLECT_EMBED_TIMEOUT, DEFLECT_LLM_TIMEOUT, deflect_answer, ) from app.domains.support.kb_service import SupportKbService query = f"{(title or '').strip()} {(body or '').strip()}".strip()[:_BODY_MAX] articles: list = [] answer: str | None = None if query: # Short HTTP timeouts: this runs synchronously on the ticket-filer's # request, so a slow backend must degrade fast (keyword KB / no AI # answer) rather than block the funnel. articles = await SupportKbService(self.db).search_published( query, limit=4, embed_timeout=DEFLECT_EMBED_TIMEOUT, ) answer = await deflect_answer(self.db, query, articles, timeout=DEFLECT_LLM_TIMEOUT) request_row = SupportServiceRequest( requester_id=person.id, origin=ORIGIN_SUPPORT_COPILOT, status=REQUEST_ACTIVE, ) self.db.add(request_row) await self.db.flush() event = SupportDeflectionEvent( person_id=person.id, service_request_id=request_row.id, query=query or "(空)", kb_article_ids=[str(a.id) for a in articles] or None, copilot_answered=bool(answer), outcome=DEFLECT_OPEN, ) self.db.add(event) await self.db.flush() return event, articles, answer async def mark_deflected( self, *, event: SupportDeflectionEvent, person: IdentityPerson, ) -> SupportDeflectionEvent: """The employee said the self-service answer solved it — a real deflection. Owner-scoped; idempotent (re-marking is a no-op).""" if event.person_id != person.id: raise AppException(code="FORBIDDEN", message="无权操作", status_code=403) if event.outcome == DEFLECT_OPEN: event.outcome = DEFLECT_DEFLECTED event.resolved_at = datetime.now(UTC) if event.service_request_id: request_row = await self.db.get(SupportServiceRequest, event.service_request_id) if request_row is not None: request_row.status = REQUEST_SELF_RESOLVED request_row.outcome = "self_solved" request_row.resolved_at = event.resolved_at return event async def link_escalation( self, *, event_id, person: IdentityPerson, ticket: SupportTicket, ) -> None: """The employee filed a ticket from a deflection session — record the escalation against the event (best-effort; bad/foreign id is ignored so a funnel-instrumentation glitch never blocks ticket creation).""" event = await self.db.get(SupportDeflectionEvent, event_id) if event is None or event.person_id != person.id: return # Idempotent: only the FIRST escalation of a still-open session counts / # links — a resent/stale deflect_event_id never rewrites the linkage. if event.outcome == DEFLECT_OPEN: event.outcome = DEFLECT_ESCALATED event.resolved_at = datetime.now(UTC) event.ticket_id = ticket.id if event.service_request_id is not None: ticket.service_request_id = event.service_request_id request_row = await self.db.get(SupportServiceRequest, event.service_request_id) if request_row is not None: request_row.status = REQUEST_ESCALATED request_row.outcome = None # ── timeline / replies ────────────────────────────────────────── async def add_reply( self, *, ticket: SupportTicket, author: IdentityPerson, body: str, is_agent: bool, trace_id: str = "", ) -> SupportTicketMessage: body = (body or "").strip() if not body: raise AppException(code="VALIDATION_ERROR", message="回复不能为空", status_code=422) if ticket.status in (STATUS_CANCELLED, STATUS_CLOSED): raise AppException( code="INVALID_STATE", message="工单已关闭，无法回复", status_code=409, ) if len(body) > _BODY_MAX: body = body[:_BODY_MAX] msg = await self._add_message( ticket, author_id=author.id, kind=MSG_COMMENT, body=body, ) now = datetime.now(UTC) if is_agent: # Adoption signal for the ai-suggest ledger — checked before the # status flip so it runs for every agent reply exactly once. await self._mark_suggestion_adoption(ticket, body) if ticket.first_response_at is None: ticket.first_response_at = now # Agent answered → ball is in the requester's court. if ticket.status in (STATUS_OPEN, STATUS_TRIAGING, STATUS_IN_PROGRESS): ticket.status = STATUS_WAITING_REQUESTER if ticket.sla_paused_at is None: ticket.sla_paused_at = now self._sync_chat( ticket, note_key="agent_reply", who=author.display_name or author.work_email or "IT", ) requester = await self._requester(ticket) if requester is not None: await self._notify( recipient=requester, template_key="support.ticket_reply.to_requester", title=f"IT 回复了你的工单 · {ticket.ticket_number}", body=body, ticket=ticket, trace_id=trace_id, context={ "ticket_number": ticket.ticket_number, "reply_body": body, }, cta_path=f"/me/support/{ticket.id}", ) else: # Requester replied → back to IT's court. A reply to a RESOLVED # ticket means the fix didn't land — reopen it (don't leave it in # "resolved but the requester is still talking" limbo). if ticket.status in (STATUS_WAITING_REQUESTER, STATUS_RESOLVED): _resume_resolution_sla(ticket, now) if ticket.status == STATUS_RESOLVED: ticket.resolved_at = None ticket.status = STATUS_IN_PROGRESS if ticket.assignee_id is not None: agent = await self.db.get(IdentityPerson, ticket.assignee_id) if agent is not None: await self._notify( recipient=agent, template_key="support.ticket_reply.to_agent", title=f"求助人回复了工单 · {ticket.ticket_number}", body=body, ticket=ticket, trace_id=trace_id, context={ "ticket_number": ticket.ticket_number, "reply_body": body, }, cta_path=f"/admin/support/tickets/{ticket.id}", ) else: await self._notify_it( template_key="support.ticket_reply.to_it", title=f"未认领工单有新回复 · {ticket.ticket_number}", body=body, ticket=ticket, trace_id=trace_id, context={ "ticket_number": ticket.ticket_number, "reply_body": body, }, ) await self._audit( action="support.ticket.reply", ticket=ticket, actor_id=author.id, trace_id=trace_id, detail={"is_agent": is_agent}, ) return msg # ── IT lifecycle actions ──────────────────────────────────────── async def add_note( self, *, ticket: SupportTicket, author: IdentityPerson | None, body: str, trace_id: str = "", ) -> SupportTicketMessage | None: """往工单时间线上加一条**记录**,不改状态、不动 SLA。 和 :meth:`add_reply` 是两件不同的事,分开的理由就是这句:回复是「球 踢给对方了」——它会把状态推到 ``waiting_requester`` 并暂停 SLA;而一条 记录(群里讨论的小结、一次交接说明)不改变谁在等谁。用 ``add_reply`` 写小结,会让一张正在被处理的工单突然显示成「等你回复」。 允许 ``author`` 为空:小结的作者是平台,不是某个人。 """ body = (body or "").strip() if not body: return None if len(body) > _BODY_MAX: body = body[:_BODY_MAX] msg = await self._add_message( ticket, author_id=author.id if author else None, kind=MSG_NOTE, body=body, ) await self._audit( action="support.ticket.note_added", ticket=ticket, actor_id=author.id if author else None, trace_id=trace_id, ) return msg async def claim( self, *, ticket: SupportTicket, agent: IdentityPerson, trace_id: str = "", ) -> SupportTicket: if ticket.status in (STATUS_CLOSED, STATUS_CANCELLED): raise AppException(code="INVALID_STATE", message="工单已关闭", status_code=409) ticket.assignee_id = agent.id if ticket.status == STATUS_OPEN: ticket.status = STATUS_TRIAGING # Claiming is acknowledgement, not an answer. First-response SLA is # met by the first substantive agent reply (or the final resolution), # so the dashboard cannot be gamed by merely picking up a ticket. await self._add_message( ticket, author_id=agent.id, kind=MSG_SYSTEM, body=f"{agent.display_name or agent.work_email} 已认领工单", ) await self._audit( action="support.ticket.claimed", ticket=ticket, actor_id=agent.id, trace_id=trace_id, ) self._sync_chat( ticket, note_key="claimed", who=agent.display_name or agent.work_email or "IT", ) requester = await self._requester(ticket) if requester is not None: # Sentinel body — system-framed copy, re-rendered per language # from the template (see the ``create`` fan-out for the why). await self._notify( recipient=requester, template_key="support.ticket_claimed", title=f"IT 已受理你的工单 · {ticket.ticket_number}", body="(see template)", ticket=ticket, trace_id=trace_id, context={ "ticket_number": ticket.ticket_number, "agent_name": agent.display_name or "IT", "ticket_title": ticket.title, }, cta_path=f"/me/support/{ticket.id}", ) return ticket async def assign( self, *, ticket: SupportTicket, assignee: IdentityPerson, actor: IdentityPerson, trace_id: str = "", ) -> SupportTicket: if ticket.status in (STATUS_CLOSED, STATUS_CANCELLED): raise AppException(code="INVALID_STATE", message="工单已关闭", status_code=409) ticket.assignee_id = assignee.id if ticket.status == STATUS_OPEN: ticket.status = STATUS_TRIAGING await self._add_message( ticket, author_id=actor.id, kind=MSG_SYSTEM, body=f"工单指派给 {assignee.display_name or assignee.work_email}", ) await self._audit( action="support.ticket.assigned", ticket=ticket, actor_id=actor.id, trace_id=trace_id, detail={"assignee_id": str(assignee.id)}, ) self._sync_chat( ticket, note_key="assigned", name=assignee.display_name or assignee.work_email or "IT", ) return ticket async def set_attributes( self, *, ticket: SupportTicket, actor: IdentityPerson, category: str | None = None, priority: str | None = None, trace_id: str = "", ) -> SupportTicket: changed = [] if category is not None and category in CATEGORIES and category != ticket.category: ticket.category = category changed.append(f"分类→{category}") if priority is not None and priority in PRIORITIES and priority != ticket.priority: ticket.priority = priority changed.append(f"优先级→{priority}") if changed: await self._add_message( ticket, author_id=actor.id, kind=MSG_SYSTEM, body="；".join(changed), ) await self._audit( action="support.ticket.updated", ticket=ticket, actor_id=actor.id, trace_id=trace_id, detail={"category": ticket.category, "priority": ticket.priority}, ) if changed: self._sync_chat( ticket, note_key="attributes", detail="、".join(changed), ) return ticket async def apply_ai_triage( self, *, ticket: SupportTicket, category: str | None, priority: str | None, trace_id: str = "", ) -> list[str]: """Apply AI auto-triage to a freshly-created ticket. Fills DEFAULTS only — never overrides an explicit employee choice: category is set only when the requester left it ``other``, priority only when left ``normal``. System message (author=None) + audit on any change.""" changed: list[str] = [] if category in CATEGORIES and category != "other" and ticket.category == "other": ticket.category = category changed.append(f"分类→{category}") if priority in PRIORITIES and priority != "normal" and ticket.priority == "normal": ticket.priority = priority changed.append(f"优先级→{priority}") if changed: await self._add_message( ticket, author_id=None, kind=MSG_SYSTEM, body="AI 自动分诊: " + "；".join(changed), ) await self._audit( action="support.ticket.ai_triaged", ticket=ticket, actor_id=None, trace_id=trace_id, detail={"category": ticket.category, "priority": ticket.priority}, ) return changed async def resolve( self, *, ticket: SupportTicket, agent: IdentityPerson, resolution: str, trace_id: str = "", ) -> SupportTicket: resolution = (resolution or "").strip() if not resolution: raise AppException(code="VALIDATION_ERROR", message="请填写解决说明", status_code=422) if ticket.status in (STATUS_CLOSED, STATUS_CANCELLED): raise AppException(code="INVALID_STATE", message="工单已关闭", status_code=409) ticket.status = STATUS_RESOLVED ticket.resolution = resolution[:_BODY_MAX] ticket.resolved_at = datetime.now(UTC) _resume_resolution_sla(ticket, ticket.resolved_at) if ticket.first_response_at is None: ticket.first_response_at = ticket.resolved_at if ticket.assignee_id is None: ticket.assignee_id = agent.id await self._add_message( ticket, author_id=agent.id, kind=MSG_RESOLUTION, body=resolution, ) await self._audit( action="support.ticket.resolved", ticket=ticket, actor_id=agent.id, trace_id=trace_id, ) self._sync_chat( ticket, note_key="resolved", who=agent.display_name or agent.work_email or "IT", ) requester = await self._requester(ticket) if requester is not None: from app.core.security import create_action_token # No-login rating magic link (mirrors provisioning's # quick-complete): the GET landing page is side-effect-free # (mail-client prefetch safe) and the POST consumes the token's # jti once — so the requester rates exactly once, from any # webview, without SSO. TTL bounds the rating window. rate_token = create_action_token( purpose="support_ticket_rate", subject_id=str(requester.id), resource_id=str(ticket.id), ttl_days=14, ) # Sentinel body — system-framed copy, re-rendered per language # from the template (see the ``create`` fan-out for the why). await self._notify( recipient=requester, template_key="support.ticket_resolved", title=f"你的工单已解决 · {ticket.ticket_number}", body="(see template)", ticket=ticket, trace_id=trace_id, context={ "ticket_number": ticket.ticket_number, "resolution": ticket.resolution or "", }, cta_path=f"/me/support/{ticket.id}", card_actions=[ CardAction( label=LocalizedText( zh="评价本次服务", en="Rate this service", ja="今回の対応を評価", ), url=f"/api/v1/support/rate?token={rate_token}", style="primary", ), ], ) await self._sync_service_request(ticket) return ticket async def reopen( self, *, ticket: SupportTicket, actor: IdentityPerson, is_agent: bool, trace_id: str = "", ) -> SupportTicket: if ticket.status not in (STATUS_RESOLVED, STATUS_CLOSED): raise AppException( code="INVALID_STATE", message="仅已解决/已关闭的工单可重新打开", status_code=409, ) ticket.status = STATUS_IN_PROGRESS ticket.resolved_at = None ticket.closed_at = None ticket.sla_paused_at = None if ( ticket.resolution_due_at is not None and _as_utc(ticket.resolution_due_at) < datetime.now(UTC) ): ticket.resolution_due_at = datetime.now(UTC) + timedelta(hours=24) await self._add_message( ticket, author_id=actor.id, kind=MSG_SYSTEM, body=f"{actor.display_name or actor.work_email} 重新打开了工单", ) await self._audit( action="support.ticket.reopened", ticket=ticket, actor_id=actor.id, trace_id=trace_id, ) self._sync_chat( ticket, note_key="reopened", who=actor.display_name or actor.work_email or "IT", ) # If the requester reopened, ping IT; if IT reopened, ping requester. if is_agent: requester = await self._requester(ticket) if requester is not None: await self._notify( recipient=requester, template_key="support.ticket_reopened.to_requester", title=f"工单已重新打开 · {ticket.ticket_number}", body=ticket.title, ticket=ticket, trace_id=trace_id, context={ "ticket_number": ticket.ticket_number, "ticket_title": ticket.title, }, cta_path=f"/me/support/{ticket.id}", ) elif ticket.assignee_id is not None: agent = await self.db.get(IdentityPerson, ticket.assignee_id) if agent is not None: await self._notify( recipient=agent, template_key="support.ticket_reopened.to_agent", title=f"求助人重新打开了工单 · {ticket.ticket_number}", body=ticket.title, ticket=ticket, trace_id=trace_id, context={ "ticket_number": ticket.ticket_number, "ticket_title": ticket.title, }, cta_path=f"/admin/support/tickets/{ticket.id}", ) else: await self._notify_it( template_key="support.ticket_reopened.to_it", title=f"工单被重新打开 · {ticket.ticket_number}", body=ticket.title, ticket=ticket, trace_id=trace_id, context={ "ticket_number": ticket.ticket_number, "ticket_title": ticket.title, }, ) await self._sync_service_request(ticket) return ticket async def close( self, *, ticket: SupportTicket, actor: IdentityPerson, is_agent: bool, trace_id: str = "", ) -> SupportTicket: if ticket.status in (STATUS_CLOSED, STATUS_CANCELLED): return ticket # idempotent ticket.status = STATUS_CLOSED ticket.closed_at = datetime.now(UTC) ticket.sla_paused_at = None await self._add_message( ticket, author_id=actor.id, kind=MSG_SYSTEM, body=f"{actor.display_name or actor.work_email} 关闭了工单", ) await self._audit( action="support.ticket.closed", ticket=ticket, actor_id=actor.id, trace_id=trace_id, ) self._sync_chat(ticket, note_key="closed") await self._sync_service_request(ticket) return ticket # ── Requester satisfaction (CSAT) ──────────────────────────────── async def rate_ticket( self, *, ticket: SupportTicket, actor: IdentityPerson, score: int, trace_id: str = "", via: str = "web", ) -> SupportTicket: """One 3-point rating per ticket, requester-only, while the ticket is resolved/closed. The caliber mirrors the Feishu helpdesk (1=不满意/2=一般/3=满意, NULL=not rated) so the two entrances stay comparable. ``via`` lands in the audit detail ('web' | 'token' | 'mcp').""" if str(ticket.requester_id) != str(actor.id): raise AppException( code="NOT_TICKET_OWNER", message="只有求助人本人可以评价", status_code=403, ) if score not in (1, 2, 3): raise AppException( code="VALIDATION_ERROR", message="评分只能是 1(不满意) / 2(一般) / 3(满意)", status_code=422, ) if ticket.status not in (STATUS_RESOLVED, STATUS_CLOSED): raise AppException( code="INVALID_STATE", message="工单解决后才能评价", status_code=409, ) # Row lock + re-check: the authed path and the magic-link path can # race the same ticket (and consume_action_jti fails OPEN on Redis # trouble, so the jti gate alone must not carry once-only). # populate_existing is LOAD-BEARING: the ticket is already in this # session's identity map (expire_on_commit=False), and without it # the FOR UPDATE re-select hands back the STALE object — the lock # blocks, then the re-check reads pre-lock state and the race is # wide open again. locked = ( await self.db.execute( select(SupportTicket) .where(SupportTicket.id == ticket.id) .with_for_update() .execution_options(populate_existing=True) ) ).scalar_one() if locked.satisfaction_score is not None: raise AppException( code="ALREADY_RATED", message="该工单已评价过", status_code=409, ) if locked.status not in (STATUS_RESOLVED, STATUS_CLOSED): # A concurrent reopen can land between the pre-lock state check # and the lock — rating an in_progress ticket would burn the # one-and-only rating on a run that isn't finished. raise AppException( code="INVALID_STATE", message="工单解决后才能评价", status_code=409, ) ticket.satisfaction_score = score ticket.satisfaction_at = datetime.now(UTC) await self._add_message( ticket, author_id=actor.id, kind=MSG_SYSTEM, body=f"满意度评价：{_SATISFACTION_LABELS.get(score, score)}", ) await self._audit( action="support.ticket.rated", ticket=ticket, actor_id=actor.id, trace_id=trace_id, detail={"score": score, "via": via}, ) return ticket async def auto_close_stale_resolved(self, *, days: int) -> int: """Close resolved tickets the requester neither confirmed nor reopened within ``days``. Bounded per run; returns the count. No notification — the requester already got the resolved card (with the rating button); auto-close is bookkeeping, not news. Processed in small SELECT-FOR-UPDATE batches, each committed before the next is selected: a single 500-row locked sweep blocks every concurrent rate/reopen on those tickets toward 1205, while a commit-inside-one-big-select releases the locks and then writes the TAIL rows from stale state (clobbering concurrent reopens). Re- selecting per batch with ``status='resolved'`` in the predicate makes both failure modes impossible. """ days = max(1, int(days)) cutoff = datetime.now(UTC) - timedelta(days=days) closed = 0 while closed < _AUTOCLOSE_MAX_PER_RUN: batch = ( ( await self.db.execute( select(SupportTicket) .where( SupportTicket.status == STATUS_RESOLVED, SupportTicket.resolved_at.is_not(None), SupportTicket.resolved_at < cutoff, ) .order_by(SupportTicket.resolved_at.asc()) .limit(_AUTOCLOSE_BATCH) .with_for_update() .execution_options(populate_existing=True) ) ) .scalars() .all() ) if not batch: break for ticket in batch: ticket.status = STATUS_CLOSED ticket.closed_at = datetime.now(UTC) await self._add_message( ticket, author_id=None, kind=MSG_SYSTEM, body=f"已解决超过 {days} 天未确认，系统自动关闭", ) await self._audit( action="support.ticket.auto_closed", ticket=ticket, actor_id=None, trace_id="", detail={"resolved_days": days}, ) await self._sync_service_request(ticket) closed += 1 # Commit the batch: releases its row locks AND advances the next # re-select past these rows (their status is no longer resolved). await self.db.commit() return closed async def find_similar_resolved( self, *, title: str, body: str = "", exclude_id: UUID | None = None, limit: int = 3, ) -> list[SupportTicket]: """Similar RESOLVED tickets for ai-suggest grounding. Recall is CJK-bigram/word shingles: SQL prefilters with OR-LIKE on the longest shingles, Python ranks by Jaccard over title+resolution. Replaces the old 'first whitespace token ILIKE' recall, which was an effective no-op for Chinese titles.""" probe = _shingles(f"{title} {body}") if not probe: return [] # Prefilter anchors on TITLE shingles with a deterministic order # (len desc, then lexicographic): pure-CJK probes are all length-2 # bigrams, so a bare len sort picked an arbitrary hash-order subset # and recall varied run to run. anchor = _shingles(title) or probe conds = [] for tok in sorted(anchor, key=lambda t: (-len(t), t))[:8]: esc = tok.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") conds.append(SupportTicket.title.ilike(f"%{esc}%", escape="\\")) rows = ( ( await self.db.execute( select(SupportTicket) .where( SupportTicket.status.in_((STATUS_RESOLVED, STATUS_CLOSED)), SupportTicket.resolution.is_not(None), or_(*conds), ) .order_by(SupportTicket.resolved_at.desc()) .limit(50) ) ) .scalars() .all() ) scored = [ (_jaccard(probe, _shingles(f"{t.title} {t.resolution or ''}")), t) for t in rows if exclude_id is None or t.id != exclude_id ] scored.sort(key=lambda p: p[0], reverse=True) return [t for score, t in scored[: max(1, limit)] if score >= 0.08] async def _mark_suggestion_adoption( self, ticket: SupportTicket, reply_body: str, ) -> None: """If a recent un-adopted ai-suggest draft for this ticket is textually close to the agent's reply, mark it adopted — the honest 'AI saved IT time' signal. Best-effort and cheap (one indexed read + difflib on ≤4k chars).""" from difflib import SequenceMatcher cutoff = datetime.now(UTC) - timedelta(hours=_ADOPTION_WINDOW_HOURS) rows = ( ( await self.db.execute( select(SupportAiSuggestion) .where( SupportAiSuggestion.ticket_id == ticket.id, SupportAiSuggestion.adopted_at.is_(None), SupportAiSuggestion.created_at >= cutoff, ) .order_by(SupportAiSuggestion.created_at.desc()) # Compare the last few drafts, not just the newest: an # agent who regenerates then sends the EARLIER draft # would otherwise never count as adoption. .limit(5) ) ) .scalars() .all() ) a = " ".join((reply_body or "").split())[:4000] if not a: return best = None best_ratio = 0.0 for row in rows: b = " ".join((row.suggestion or "").split())[:4000] if not b: continue ratio = SequenceMatcher(None, a, b).ratio() if ratio > best_ratio: best, best_ratio = row, ratio if best is not None and best_ratio >= _ADOPTION_MIN_RATIO: best.adopted_at = datetime.now(UTC) best.adopted_similarity = round(best_ratio, 3) async def expire_stale_deflections(self, *, days: int = 3) -> dict[str, int]: """Sweep deflection sessions the employee never resolved (``outcome=open`` older than ``days``) and **escalate them to a real ticket** so a human sees the request. Until 2026-08-25 this swept them to ``expired`` / ``status='abandoned'`` and stopped there: no ticket, no assignment, no notification to anyone. An employee who asked for help and did not click "this solved it" had their request silently deleted after three days. Production evidence for why "didn't click" must not be read as "solved": of the 10 deflection sessions ever recorded, **zero** ended in ``deflected`` — nobody has ever pressed the confirm button — while 6 were swept to ``expired`` and vanished. The abandon path was not an edge case, it was the majority path. One of those six was a US engineer asking twice why his AWS account had disappeared; both requests expired unseen, and he mailed IT six days later to say nobody had answered. He was right. ``outcome`` stays :data:`DEFLECT_EXPIRED` — that field records what the EMPLOYEE did (they never chose), which is what the KB-gap job and the funnel analytics read it for. What changes is what *we* do about it: the service request becomes ``escalated`` with a ticket attached instead of ``abandoned`` with nothing. Bounded per run. Returns per-outcome counts. """ days = max(1, int(days)) cutoff = datetime.now(UTC) - timedelta(days=days) rows = ( ( await self.db.execute( select(SupportDeflectionEvent) .where( SupportDeflectionEvent.outcome == DEFLECT_OPEN, SupportDeflectionEvent.created_at < cutoff, ) .order_by(SupportDeflectionEvent.created_at.asc()) .limit(1000) ) ) .scalars() .all() ) escalated = 0 unescalated = 0 for event in rows: event.outcome = DEFLECT_EXPIRED event.resolved_at = datetime.now(UTC) request_row = ( await self.db.get(SupportServiceRequest, event.service_request_id) if event.service_request_id else None ) ticket = await self._auto_escalate_deflection(event) if ticket is not None: event.ticket_id = ticket.id if request_row is not None: # ``escalated`` with a linked ticket — the same terminal # state ``link_escalation`` writes when the employee files # one themselves. The operations analytics derives # ``agent_escalated`` from ticket linkage, so these rows # now leave the ``agent_abandoned`` bucket honestly rather # than by relabelling. request_row.status = REQUEST_ESCALATED request_row.outcome = None request_row.resolved_at = None escalated += 1 else: # Escalation genuinely not possible (no question text to act # on, or the requester has since left). Fall back to the old # terminal state rather than leaving the row open forever. if request_row is not None: request_row.status = REQUEST_ABANDONED request_row.outcome = "unknown" request_row.resolved_at = event.resolved_at unescalated += 1 return { "scanned": len(rows), "escalated": escalated, "unescalated": unescalated, } async def _auto_escalate_deflection( self, event: SupportDeflectionEvent, ) -> SupportTicket | None: """File a ticket on the employee's behalf for an unresolved deflection session. Returns the ticket, or ``None`` when escalation is not possible or failed. Never raises: an escalation failure must not abort the sweep and strand the remaining sessions, and the caller has already decided the session is terminal either way. """ query = (event.query or "").strip() if not query or query == "(空)": # Nothing actionable was ever typed — a ticket would be an empty # ask, and ``create`` rejects a blank body anyway. return None requester = await self.db.get(IdentityPerson, event.person_id) if ( requester is None or requester.deleted_at is not None or requester.employment_status in ("left", "leaver_pending") ): # A departing employee's open questions belong to the offboarding # pipeline, not to a fresh ticket filed days after the fact. return None # ``query`` is the concatenated "{title} {body}" the funnel stored, so # the title is reconstructed from its head rather than stored twice. first_line = next( (ln.strip() for ln in query.splitlines() if ln.strip()), query, ) title = first_line[:80].rstrip() try: return await self.create( requester=requester, title=title, body=query, category="other", priority="normal", source=SOURCE_AI, service_request_id=event.service_request_id, # The EMPLOYEE wanted help; the SYSTEM filed the ticket. Both # halves are recorded, so "who asked" and "who typed it in" # never collapse into one attribution. intent_initiator=INITIATOR_EMPLOYEE, creation_actor=INITIATOR_SYSTEM, creation_mode=CREATION_AGENT_AUTOMATIC, origin_channel=ORIGIN_SUPPORT_COPILOT, # Deliberately NOT set: the employee never confirmed this # ticket, and stamping a confirmation would launder a # system-filed ticket into an employee-confirmed one. employee_confirmed_at=None, trace_id="support.deflection_auto_escalate", ) except Exception as exc: # noqa: BLE001 logger.warning( "deflection auto-escalation failed for event %s: %s", event.id, exc, ) return None async def cancel( self, *, ticket: SupportTicket, requester: IdentityPerson, trace_id: str = "", ) -> SupportTicket: if ticket.status not in OPEN_STATUSES: raise AppException( code="INVALID_STATE", message="仅未完成的工单可撤回", status_code=409, ) ticket.status = STATUS_CANCELLED ticket.closed_at = datetime.now(UTC) ticket.sla_paused_at = None await self._add_message( ticket, author_id=requester.id, kind=MSG_SYSTEM, body="求助人撤回了工单", ) await self._audit( action="support.ticket.cancelled", ticket=ticket, actor_id=requester.id, trace_id=trace_id, ) self._sync_chat(ticket, note_key="cancelled") await self._sync_service_request(ticket) return ticket # ── reads ─────────────────────────────────────────────────────── async def get( self, ticket_id: UUID, *, for_update: bool = False, ) -> SupportTicket | None: # ``for_update`` row-locks the ticket so concurrent admin # transitions (two agents / a double-click resolving the same # ticket) serialize and re-check ``status`` atomically — otherwise # both flip open→resolved and each fire a ``ticket_resolved`` # notification (duplicate pings) / overwrite ``assignee_id`` twice. if for_update: return await self.db.scalar( select(SupportTicket).where(SupportTicket.id == ticket_id).with_for_update() ) return await self.db.get(SupportTicket, ticket_id) async def list_messages(self, ticket_id: UUID) -> list[SupportTicketMessage]: rows = ( ( await self.db.execute( select(SupportTicketMessage) .where(SupportTicketMessage.ticket_id == ticket_id) .order_by(SupportTicketMessage.created_at.asc()) ) ) .scalars() .all() ) return list(rows) async def list_for_requester( self, *, person_ids: list[UUID], status: str | None = None, page: int = 1, page_size: int = 20, ) -> tuple[list[SupportTicket], int]: stmt = select(SupportTicket).where(SupportTicket.requester_id.in_(person_ids)) if status == "open": stmt = stmt.where(SupportTicket.status.in_(OPEN_STATUSES)) elif status: stmt = stmt.where(SupportTicket.status == status) return await self._paginate(stmt, page, page_size) async def list_queue( self, *, status: str | None = None, assignee_id: UUID | None = None, category: str | None = None, priority: str | None = None, q: str | None = None, page: int = 1, page_size: int = 20, ) -> tuple[list[SupportTicket], int]: stmt = select(SupportTicket) if status == "open": stmt = stmt.where(SupportTicket.status.in_(OPEN_STATUSES)) elif status == "resolved": stmt = stmt.where(SupportTicket.status.in_(RESOLVED_STATUSES)) elif status == "all": pass elif status: stmt = stmt.where(SupportTicket.status == status) if assignee_id is not None: stmt = stmt.where(SupportTicket.assignee_id == assignee_id) if category: stmt = stmt.where(SupportTicket.category == category) if priority: stmt = stmt.where(SupportTicket.priority == priority) qc = (q or "").strip()[:100] if qc: esc = qc.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") like = f"%{esc}%" stmt = stmt.where( or_( SupportTicket.title.ilike(like, escape="\\"), SupportTicket.ticket_number.ilike(like, escape="\\"), ) ) return await self._paginate(stmt, page, page_size) async def _paginate( self, stmt, page: int, page_size: int, ) -> tuple[list[SupportTicket], int]: page = max(1, page) page_size = max(1, min(page_size, 100)) total = ( await self.db.execute(select(func.count()).select_from(stmt.subquery())) ).scalar() or 0 rows = ( ( await self.db.execute( stmt.order_by(SupportTicket.created_at.desc()) .offset((page - 1) * page_size) .limit(page_size) ) ) .scalars() .all() ) return list(rows), int(total) async def queue_stats(self) -> dict[str, int]: """Per-status counts for the IT workbench header.""" rows = ( await self.db.execute( select(SupportTicket.status, func.count()).group_by(SupportTicket.status) ) ).all() counts = {status: int(n) for status, n in rows} counts["open_total"] = sum(counts.get(s, 0) for s in OPEN_STATUSES) counts["resolved_total"] = sum(counts.get(s, 0) for s in RESOLVED_STATUSES) counts["total_total"] = int(sum(int(n) for _status, n in rows)) return counts