"""工单群的收尾:结了的群解散掉,没人管的群问一句。 ## 为什么需要一个定时任务 群是**不可撤销地**建出来的,而且会一直留在每个成员的会话列表里。工单结了之后 那个房间就只是噪音 —— 但没有人会想起来去删它。第一批工单群上线一周后,负责人的 飞书里就会有一排「IT 求助 · …」,而其中大多数早就结束了。 ## 两件事,两种温度 * **结了的群 → 解散**(保留 N 天)。删之前保证工单上有一段过程记录:房间的聊天 记录跟着它一起消失,而那段过程是这条链路最值钱的产物。 * **没人管的群 → 问一句**。刻意**不自动关单**:自动关掉一张还开着的工单,会让 「已关闭」这个状态不再代表任何东西。问一句并给出一条出路(「解决了 …」)就够。 ## 失败怎么处理 单个房间失败(飞书抖动、权限变了)**不算这次运行失败**:状态留在 ``open``,下一 次扫描重试。但**全部都失败**要抛 —— 那不是抖动,那是权限没了或者凭据坏了,而 一个安静地每天什么都做不成的清理任务,比没有这个任务更糟。 """ from __future__ import annotations import logging from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import and_, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import aliased from app.core.locale import Lang from app.scheduled_jobs.registry import register_scheduled_job logger = logging.getLogger(__name__) #: 工单结了之后保留几天。0 = 永不自动解散(仍然可以手动解散)。 SETTING_RETENTION_DAYS = "assistant.ticket_chat_retention_days" DEFAULT_RETENTION_DAYS = 7 #: 群里多久没动静就问一句。0 = 不问。 SETTING_IDLE_NUDGE_DAYS = "assistant.ticket_chat_idle_nudge_days" DEFAULT_IDLE_NUDGE_DAYS = 14 #: 一次最多处理多少个房间。每个都是一到三次飞书往返,而这个任务不该独占 #: worker —— 剩下的下一次接着来(状态在库里,不在这次运行里)。 _BATCH = 50 @register_scheduled_job( key="assistant.ticket_chat_sweep", display_name="助手 · 工单群收尾", description=( "结了的工单群按保留期解散(解散前保证工单上有过程记录 —— 群的聊天记录" "跟着它一起消失);还开着但长期没动静的群里问一句「这件事还要不要」。" "刻意不自动关单:自动关掉一张还开着的工单,会让「已关闭」不再代表任何" "东西。单个房间失败不算这次失败(下一次重试),全部失败才抛。" ), weight="light", ) async def sweep_ticket_chats( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.assistant import ticket_chat from app.core.locale import DEFAULT_LANG, resolve_lang from app.domains.support.models import SupportTicket from app.identity.person_models import IdentityPerson from app.integrations.service import get_platform_setting cfg = config or {} # 🔴 **config 里带这两个键就抛。** 2026-09-08 的 #3520 正是反过来的那一半: # 播种进 `scheduled_jobs.config` 的一把钥匙赢过了平台设置,于是设置页上的 # 改动被静默无视一周,而排查它只能靠读源码。那次的修法是「填了就抛,并在 # 报错里给出该改的地方」;这一行照办 —— 一个被安静忽略的 config 键,表现 # 同样是「我改了配置但行为没变」。 stray = sorted({"retention_days", "idle_nudge_days"} & set(cfg)) if stray: raise ValueError( "assistant.ticket_chat_sweep: 不要把 " f"{', '.join(stray)} 写进 scheduled_jobs.config —— " f"改平台设置 {SETTING_RETENTION_DAYS} / {SETTING_IDLE_NUDGE_DAYS}", ) retention = _clamp( await get_platform_setting( SETTING_RETENTION_DAYS, DEFAULT_RETENTION_DAYS, session=db, ), DEFAULT_RETENTION_DAYS, ) idle_days = _clamp( await get_platform_setting( SETTING_IDLE_NUDGE_DAYS, DEFAULT_IDLE_NUDGE_DAYS, session=db, ), DEFAULT_IDLE_NUDGE_DAYS, ) now = datetime.now(UTC) rows = list((await db.execute( _due_rooms(now, retention=retention, idle_days=idle_days) )).scalars().all()) considered = 0 disbanded = 0 failed = 0 nudged = 0 attempted = 0 # 「本该清掉、但为了保住过程记录先留着」的数量。单独一个计数,因为它和 # 「没到期」长得一样(都是 disbanded == 0),而原因完全不同:一个是策略, # 一个是模型或权限出了问题。合在一起就再也看不出「这个任务其实什么都没 # 干成」。 kept_for_record = 0 for chat in rows: ticket = await db.get(SupportTicket, chat.ticket_id) if ticket is None: # 工单被硬删了,房间还在。解散它 —— 这个群讲的那件事已经不存在。 attempted += 1 considered += 1 if await _disband_orphan(db, chat=chat): disbanded += 1 else: failed += 1 continue considered += 1 requester = await db.get(IdentityPerson, ticket.requester_id) lang = resolve_lang(requester) if requester else DEFAULT_LANG if ticket.status in ("closed", "cancelled"): if retention <= 0: continue settled_at = ticket.closed_at or ticket.updated_at if settled_at is None: continue if _as_utc(settled_at) > now - timedelta(days=retention): continue # 删之前先保证台账上有过程记录:房间的聊天记录跟着它一起消失。 # **记录做不出来就留着房间** —— 记录比房间干净重要。 if not await _ensure_record(db, chat=chat, ticket=ticket, lang=lang): kept_for_record += 1 continue attempted += 1 if await ticket_chat.disband( db, chat=chat, ticket=ticket, reason="retention", lang=lang, ): disbanded += 1 else: failed += 1 continue if idle_days <= 0: continue last = chat.last_activity_at or chat.created_at if last is None or _as_utc(last) > now - timedelta(days=idle_days): continue if await ticket_chat.nudge_idle( db, chat=chat, ticket=ticket, days=idle_days, lang=lang, ): # 问过就算一次活动 —— 否则每天都会再问一遍。 ticket_chat.touch(chat) nudged += 1 await db.commit() if attempted and failed == attempted: # 全部失败不是抖动:是权限没了、凭据坏了,或者应用被移出了那些群。 # 一个安静地每天什么都做不成的清理任务,比没有这个任务更糟。 raise RuntimeError( f"assistant.ticket_chat_sweep: {failed}/{attempted} 个群都没能解散 —— " "检查飞书 im:chat 权限与应用凭据", ) return { "considered": considered, "disbanded": disbanded, "nudged": nudged, "failed": failed, "kept_for_record": kept_for_record, "retention_days": retention, "idle_nudge_days": idle_days, } def _due_rooms(now: datetime, *, retention: int, idle_days: int): """这一轮**可能需要动**的房间。 🔴 **到期判断必须在 SQL 里,不能在 Python 里。** 取「最老的 50 个开着的 房间」再逐个判,会在一个很普通的稳定状态下**永久卡死**:一张长期开着的 工单(跨月的采购、等外部厂商的那种)对应的房间会一直排在 ``created_at`` 的最前面,而它永远不该被动。这样的房间攒到 50 个,这个任务就再也走不到 更新的房间 —— 而它每天都返回 ``considered=50, disbanded=0``,读起来完全 像「没有到期的」。500 人的公司里同时开着 50 张长命工单不是极端情形。 三种「可能需要动」: * 工单已经不在了(硬删)—— 这个群讲的那件事已经不存在; * 工单结了,且过了保留期; * 工单还开着,且房间超过 ``idle_days`` 没动静。 ``retention``/``idle_days`` 为 0 时对应的那一支直接不出现在条件里(0 = 关掉这项),于是查询只剩孤儿房间 —— 而不是把 50 个名额浪费在必然 ``continue`` 的行上。 """ from app.assistant.models import CHAT_STATE_OPEN, AssistantTicketChat from app.domains.support.models import SupportTicket # 外连一个别名:``SupportTicket`` 在同一次查询里没有别的用途,但取别名让 # 「工单没了」这一支能用 ``ticket.id IS NULL`` 表达,而不用第二次查询。 ticket = aliased(SupportTicket) settled = ticket.status.in_(("closed", "cancelled")) # 存的是 UTC naive(``DateTime(timezone=True)`` 在 MySQL 上就是 DATETIME), # 所以比较值要脱掉 tzinfo —— 带 offset 的值在驱动那一层的行为是另一个陷阱。 naive = now.replace(tzinfo=None) clauses: list[Any] = [ticket.id.is_(None)] if retention > 0: clauses.append(and_( settled, func.coalesce(ticket.closed_at, ticket.updated_at) <= naive - timedelta(days=retention), )) if idle_days > 0: clauses.append(and_( ticket.id.is_not(None), ~settled, func.coalesce( AssistantTicketChat.last_activity_at, AssistantTicketChat.created_at, ) <= naive - timedelta(days=idle_days), )) return ( select(AssistantTicketChat) .outerjoin(ticket, ticket.id == AssistantTicketChat.ticket_id) .where(AssistantTicketChat.state == CHAT_STATE_OPEN, or_(*clauses)) .order_by(AssistantTicketChat.created_at) .limit(_BATCH) ) async def _ensure_record( db: AsyncSession, *, chat: Any, ticket: Any, lang: Lang, ) -> bool: """工单上至少要有一段过程记录,再删房间。返回**能不能删了**。 群的聊天记录跟着解散一起消失,而那段过程是这条链路最值钱的产物(交接、 知识库语料、复盘都读它)。已经有记录就不再花一次模型调用。 三种「可以删」:已经有记录、房间里压根没人说过话(没有东西会丢)、这一次 刚把小结写上去。剩下的一律返回 ``False`` —— 留着房间,下一次再试。 """ from app.assistant import room_history, room_summary from app.domains.support.models import MSG_NOTE, SupportTicketMessage from app.domains.support.service import SupportService existing = (await db.execute( select(SupportTicketMessage.id) .where( SupportTicketMessage.ticket_id == ticket.id, SupportTicketMessage.kind == MSG_NOTE, ) .limit(1) )).scalars().first() if existing is not None: return True transcript = await room_history.read_room(db, chat_id=chat.feishu_chat_id) if transcript is None: # 读不到这个房间(权限变了 / 飞书抖动)。**留着** —— 屋里的人还能看到 # 那段历史,而我们没法把它保下来。这一格计进 ``kept_for_record``, # 于是「这个任务其实什么都没干成」在运行结果里看得出来。 logger.warning( "assistant: cannot read chat %s — keeping it until its history " "can be preserved", chat.feishu_chat_id, ) return False if not transcript.human_messages: # 一个没人说过话的房间,删掉不丢任何东西。 return True summary = await room_summary.summarize( db, transcript=transcript, ticket_title=ticket.title, lang=lang, ) if summary is None: # 模型不可用:**不写占位**(见 ``room_summary`` 的模块头),而是把这个 # 房间留到下一次 —— 记录比房间干净重要。 logger.info( "assistant: keeping chat %s until its discussion can be recorded", chat.feishu_chat_id, ) return False await SupportService(db).add_note( ticket=ticket, author=None, body=summary.render(lang), ) return True async def _disband_orphan(db: AsyncSession, *, chat: Any) -> bool: """工单没了的房间。没有工单号可念,所以只发一句然后删。""" from app.assistant import channel, ticket_chat from app.assistant.channel import ReplyTarget client = await channel.build_client(db) if client is None: return False try: await channel.send_text( db, to=ReplyTarget.chat(chat.feishu_chat_id), text="这个群对应的工单已经不存在了,群随后解散。", ) await client.disband_chat(chat_id=chat.feishu_chat_id) except Exception as exc: # noqa: BLE001 logger.warning("assistant: orphan chat cleanup failed: %s", exc) return False ticket_chat.mark_closed(chat, reason="ticket_gone") # 同一行审计:删掉的东西和手动解散是同一个(一整段聊天记录),只是没有 # 操作人 —— ``actor_type`` 因此是 ``system``。 await ticket_chat.audit_disband(db, chat=chat, reason="ticket_gone") return True def _as_utc(value: datetime) -> datetime: """naive 的时间戳当 UTC 读 —— MySQL 的 DATETIME 回来没有时区。""" return value if value.tzinfo else value.replace(tzinfo=UTC) def _clamp(setting: Any, default: int) -> int: """平台设置 → 一个 0..365 的整数,读不出来就用代码默认。 **没有 config 这一档**:这两个阈值只有一个来源(见上面那段关于 #3520 的 注释)。多一个来源就多一个「我改了配置但行为没变」。 """ if setting is not None: try: return max(0, min(int(setting), 365)) except (TypeError, ValueError): pass return default __all__ = [ "DEFAULT_IDLE_NUDGE_DAYS", "DEFAULT_RETENTION_DAYS", "SETTING_IDLE_NUDGE_DAYS", "SETTING_RETENTION_DAYS", "sweep_ticket_chats", ]