"""工单域接口（M2）：升级、指派、办理、验收、回流、通知、事项关系。 一段流程长什么样： 任务卡住了 -> POST /items/{task_id}/promote {assignee_id} 任务变 blocked，生成一条工单（open） -> 被指派人 POST /items/{ticket_id}/ticket/accept accepted -> POST /items/{ticket_id}/ticket/start in_progress -> POST /items/{ticket_id}/ticket/submit pending_verify（提需求的人收到通知） -> POST /items/{ticket_id}/ticket/verify {approve} done，原任务自动 blocked -> todo 驳回走同一个入口：verify {approve: false, note: "..."} -> 回到 in_progress，reject_count + 1。 每一步都写 activity；通知只是提醒，状态永远以 item 为准。 """ from __future__ import annotations import uuid from datetime import datetime, timezone from typing import Annotated, Any from fastapi import APIRouter, Query from pydantic import BaseModel, Field from sqlalchemy import Select, and_, or_, select, update from sqlalchemy.ext.asyncio import AsyncSession from ..audit import record as audit_record from ..deps import MemberDep, SessionDep, TenantCtx from ..domain.ticket import ( CANCELABLE, NOW_FIELD, ensure_can_act, returned_task_status, target_status, ) from ..errors import AppError, ErrorCode from ..models import Member from ..models_task import Activity, Item from ..models_ticket import ItemLink, Notification, TicketExt from ..schemas import Pagination, encode_cursor, ok from ..serializers import item_out, tags_of router = APIRouter(prefix="/workspaces/{ws_id}", tags=["工单"]) notify_router = APIRouter(prefix="/workspaces/{ws_id}/notifications", tags=["通知"]) MAX_REJECT_NOTE = 500 # -- 入参 --------------------------------------------------------------- class PromoteIn(BaseModel): assignee_id: uuid.UUID title: str | None = Field(default=None, max_length=200) body_md: str | None = Field(default=None, max_length=20000) due_at: datetime | None = None priority: int | None = Field(default=None, ge=0, le=3) version: int | None = Field(default=None, description="原任务的版本，给了就做乐观锁") class ReassignIn(BaseModel): assignee_id: uuid.UUID class VerifyIn(BaseModel): approve: bool note: str = Field(default="", max_length=MAX_REJECT_NOTE) resolution_md: str = Field( default="", max_length=20000, description="验收通过时写的结论，M3 用它长知识" ) class ActionIn(BaseModel): note: str = Field(default="", max_length=MAX_REJECT_NOTE) version: int | None = None class LinkIn(BaseModel): to_item_id: uuid.UUID link_type: str = "related" # -- 内部工具 ----------------------------------------------------------- async def _load_item( session: AsyncSession, ws_id: uuid.UUID, item_id: uuid.UUID, kinds: tuple[str, ...] ) -> Item: item = await session.get(Item, item_id) if item is None or item.workspace_id != ws_id or item.kind not in kinds: # 跨租户 / 不存在 / 类型不符：对外一律"不存在" raise AppError(ErrorCode.ITEM_NOT_FOUND, "事项不存在") if item.deleted_at is not None: raise AppError(ErrorCode.ITEM_NOT_FOUND, "事项不存在") return item async def _load_ext(session: AsyncSession, item_id: uuid.UUID) -> TicketExt: ext = await session.get(TicketExt, item_id) if ext is None: raise AppError(ErrorCode.TICKET_NOT_FOUND, "这不是一条工单") return ext async def _active_member_id(session: AsyncSession, ws_id: uuid.UUID, user_id: uuid.UUID) -> None: exists = ( await session.execute( select(Member.id).where( Member.workspace_id == ws_id, Member.user_id == user_id, Member.status == "active", ) ) ).first() if exists is None: raise AppError(ErrorCode.BAD_REQUEST, "被指派的人不在这个工作区里") async def _log( session: AsyncSession, item: Item, actor_id: uuid.UUID | None, action: str, payload: dict | None = None, ) -> None: session.add( Activity( workspace_id=item.workspace_id, item_id=item.id, actor_id=actor_id, action=action, payload=payload or {}, ) ) async def _notify( session: AsyncSession, *, ws_id: uuid.UUID, user_id: uuid.UUID, kind: str, item_id: uuid.UUID | None, title: str, body: str = "", ) -> None: session.add( Notification( workspace_id=ws_id, user_id=user_id, kind=kind, item_id=item_id, title=title[:120], body=body[:300], ) ) def _ticket_ext_out(ext: TicketExt) -> dict[str, Any]: return { "requester_id": str(ext.requester_id), "assignee_id": str(ext.assignee_id), "parent_item_id": str(ext.parent_item_id) if ext.parent_item_id else None, "accepted_at": ext.accepted_at, "submitted_at": ext.submitted_at, "verified_at": ext.verified_at, "verified_by": str(ext.verified_by) if ext.verified_by else None, "reject_count": ext.reject_count, "resolution_md": ext.resolution_md, } async def _item_out(session: AsyncSession, ws_id: uuid.UUID, item: Item) -> dict[str, Any]: """单条序列化。标签必须一起带上——前端把 `item.tags` 当数组用， 少这个字段整页会白屏（M2 首版就是这么炸的）。""" grouped = await tags_of(session, ws_id, [item.id]) return item_out(item, grouped.get(item.id)) async def _items_out( session: AsyncSession, ws_id: uuid.UUID, items: list[Item] ) -> list[dict[str, Any]]: grouped = await tags_of(session, ws_id, [i.id for i in items]) return [item_out(i, grouped.get(i.id)) for i in items] async def _ticket_out( session: AsyncSession, ws_id: uuid.UUID, ticket: Item, ext: TicketExt ) -> dict: return ok({**await _item_out(session, ws_id, ticket), "ticket": _ticket_ext_out(ext)}) async def _patch_item( session: AsyncSession, item: Item, values: dict[str, Any], *, expect_version: int | None, ) -> None: """带版本的 UPDATE：给了 version 就必须匹配，否则 409 让客户端刷新。""" where = [Item.id == item.id, Item.workspace_id == item.workspace_id] if expect_version is not None: where.append(Item.version == expect_version) result = await session.execute( update(Item) .where(*where) .values(**values, version=Item.version + 1, updated_at=datetime.now(timezone.utc)) ) if result.rowcount == 0: await session.refresh(item) raise AppError( ErrorCode.VERSION_CONFLICT, "这条工单刚被改过，请刷新后重试", details={"current_version": item.version}, ) await session.refresh(item) # -- 升级 --------------------------------------------------------------- @router.post("/items/{item_id}/promote", summary="把任务升级成工单") async def promote(item_id: uuid.UUID, body: PromoteIn, ctx: MemberDep, session: SessionDep) -> dict: task = await _load_item(session, ctx.ws_id, item_id, ("task",)) if task.status in ("done", "canceled"): raise AppError( ErrorCode.ITEM_STATUS_INVALID, f"「{task.status}」的任务不能再升级成工单", details={"status": task.status}, ) existing = ( await session.execute( select(TicketExt.item_id).where( TicketExt.workspace_id == ctx.ws_id, TicketExt.parent_item_id == task.id ) ) ).first() if existing is not None: raise AppError(ErrorCode.TICKET_ALREADY_EXISTS, "这条任务已经升级过工单了") await _active_member_id(session, ctx.ws_id, body.assignee_id) ticket = Item( workspace_id=ctx.ws_id, kind="ticket", title=(body.title or task.title)[:200], body_md=body.body_md if body.body_md is not None else task.body_md, status="open", priority=body.priority if body.priority is not None else task.priority, creator_id=ctx.user.id, assignee_id=body.assignee_id, due_at=body.due_at or task.due_at, ) session.add(ticket) await session.flush() ext = TicketExt( item_id=ticket.id, workspace_id=ctx.ws_id, requester_id=ctx.user.id, assignee_id=body.assignee_id, parent_item_id=task.id, ) session.add(ext) await session.flush() # 原任务确实卡住了，进 blocked；不能静默把用户的待办改成别的状态 await _patch_item(session, task, {"status": "blocked"}, expect_version=body.version) await _log(session, task, ctx.user.id, "promoted_to_ticket", {"ticket_id": str(ticket.id)}) await _log(session, ticket, ctx.user.id, "ticket_created", {"parent_item_id": str(task.id)}) await _notify( session, ws_id=ctx.ws_id, user_id=body.assignee_id, kind="ticket_assigned", item_id=ticket.id, title=f"你被指派了一条工单：{ticket.title}", body=f"{ctx.user.nickname or '同事'} 请你处理，完成后提交验收", ) await audit_record( session, workspace_id=ctx.ws_id, actor_user_id=ctx.user.id, action="ticket.promote", target_type="item", target_id=str(ticket.id), payload={"parent_item_id": str(task.id)}, ) await session.flush() task_out, ticket_out = await _items_out(session, ctx.ws_id, [task, ticket]) return ok({"task": task_out, "ticket": {**ticket_out, "ticket": _ticket_ext_out(ext)}}) # -- 办理/验收 ---------------------------------------------------------- async def _draft_knowledge( session: AsyncSession, ctx: TenantCtx, ticket: Item, ext: TicketExt, resolution_md: str, now: datetime, ) -> None: """验收通过时自动整理一份知识草稿（M3 闭环沉淀）。 只落 draft 不发布：自动整理的东西得有人扫一眼才敢进团队检索。 同一张工单只沉淀一次（重复点验收、历史数据补跑都不会重复产出）。 """ from ..services.knowledge import create_draft_from_source, find_doc_by_source if await find_doc_by_source(session, ctx.ws_id, ticket.id) is not None: return doc_item, _doc = await create_draft_from_source( session, ws_id=ctx.ws_id, source=ticket, author_id=ctx.user.id, generated_by="rule", resolution_md=resolution_md or ext.resolution_md or "", doc_type="postmortem", ) await _log( session, doc_item, ctx.user.id, "knowledge_created", {"source_item_id": str(ticket.id), "generated_by": "rule"}, ) async def _advance( ctx: TenantCtx, session: AsyncSession, item_id: uuid.UUID, *, action: str, note: str = "", approve: bool | None = None, resolution_md: str = "", version: int | None = None, ) -> dict: ticket = await _load_item(session, ctx.ws_id, item_id, ("ticket",)) ext = await _load_ext(session, ticket.id) target = target_status(action, ticket.status) ensure_can_act( action=action, actor_id=ctx.user.id, role=ctx.role, assignee_id=ext.assignee_id, requester_id=ext.requester_id, ) now = datetime.now(timezone.utc) previous = ticket.status values: dict[str, Any] = {"status": target} if target == "done": values["completed_at"] = now ext.verified_at = now ext.verified_by = ctx.user.id if resolution_md: ext.resolution_md = resolution_md if target == "in_progress" and action == "reject": ext.reject_count = ext.reject_count + 1 stamp_field = NOW_FIELD.get(target) if stamp_field and action != "reject": setattr(ext, stamp_field, now) await _patch_item(session, ticket, values, expect_version=version) payload: dict[str, Any] = {"action": action, "from": previous, "to": target} if note: payload["note"] = note await _log(session, ticket, ctx.user.id, f"ticket_{action}", payload) # 通知对家：验收找提需求的人，其余找办理人 if action == "submit": await _notify( session, ws_id=ctx.ws_id, user_id=ext.requester_id, kind="ticket_submitted", item_id=ticket.id, title=f"工单待你验收：{ticket.title}", body=note or "办理人已提交，请确认结果", ) elif action == "approve": await _notify( session, ws_id=ctx.ws_id, user_id=ext.assignee_id, kind="ticket_approved", item_id=ticket.id, title=f"工单已通过验收：{ticket.title}", body=note or "谢谢，这事结了", ) elif action == "reject": await _notify( session, ws_id=ctx.ws_id, user_id=ext.assignee_id, kind="ticket_rejected", item_id=ticket.id, title=f"工单被退回：{ticket.title}", body=note or "请按反馈再改一版", ) # 验收通过 -> 原任务闭环回流 returned = returned_task_status(target) if returned and ext.parent_item_id: parent = await session.get(Item, ext.parent_item_id) if parent is not None and parent.workspace_id == ctx.ws_id and parent.status == "blocked": parent.status = returned parent.version += 1 parent.updated_at = now await _log( session, parent, ctx.user.id, "ticket_done", {"ticket_id": str(ticket.id), "status": returned}, ) await _notify( session, ws_id=ctx.ws_id, user_id=parent.creator_id, kind="task_unblocked", item_id=parent.id, title=f"任务可以继续了：{parent.title}", body="它对应的工单已验收通过", ) # 验收通过 -> 顺手整理一份知识草稿（闭环沉淀）。 # 只落草稿不发布：自动整理的东西得有人扫一眼才敢进团队检索。 if action == "approve": await _draft_knowledge(session, ctx, ticket, ext, resolution_md, now) await session.flush() return await _ticket_out(session, ctx.ws_id, ticket, ext) @router.post("/items/{item_id}/ticket/accept", summary="接受工单") async def accept(item_id: uuid.UUID, ctx: MemberDep, session: SessionDep) -> dict: return await _advance(ctx, session, item_id, action="accept") @router.post("/items/{item_id}/ticket/start", summary="开始办理") async def start(item_id: uuid.UUID, ctx: MemberDep, session: SessionDep) -> dict: return await _advance(ctx, session, item_id, action="start") @router.post("/items/{item_id}/ticket/submit", summary="提交验收") async def submit(item_id: uuid.UUID, body: ActionIn, ctx: MemberDep, session: SessionDep) -> dict: return await _advance( ctx, session, item_id, action="submit", note=body.note, version=body.version ) @router.post("/items/{item_id}/ticket/verify", summary="验收：通过或驳回") async def verify(item_id: uuid.UUID, body: VerifyIn, ctx: MemberDep, session: SessionDep) -> dict: return await _advance( ctx, session, item_id, action="approve" if body.approve else "reject", note=body.note, approve=body.approve, resolution_md=body.resolution_md, ) @router.post("/items/{item_id}/ticket/cancel", summary="取消工单") async def cancel(item_id: uuid.UUID, body: ActionIn, ctx: MemberDep, session: SessionDep) -> dict: return await _advance( ctx, session, item_id, action="cancel", note=body.note, version=body.version ) @router.post("/items/{item_id}/ticket/reassign", summary="改派给同事") async def reassign( item_id: uuid.UUID, body: ReassignIn, ctx: MemberDep, session: SessionDep ) -> dict: ticket = await _load_item(session, ctx.ws_id, item_id, ("ticket",)) ext = await _load_ext(session, ticket.id) ensure_can_act( action="reassign", actor_id=ctx.user.id, role=ctx.role, assignee_id=ext.assignee_id, requester_id=ext.requester_id, ) if ticket.status not in CANCELABLE: raise AppError( ErrorCode.TICKET_STATUS_INVALID, f"「{ticket.status}」的工单不能改派", details={"allowed_from": list(CANCELABLE)}, ) if body.assignee_id == ext.assignee_id: return await _ticket_out(session, ctx.ws_id, ticket, ext) await _active_member_id(session, ctx.ws_id, body.assignee_id) await _patch_item( session, ticket, {"status": "open", "assignee_id": body.assignee_id, "completed_at": None}, expect_version=None, ) ext.assignee_id = body.assignee_id ext.accepted_at = None ext.submitted_at = None await _log(session, ticket, ctx.user.id, "ticket_reassigned", {"to": str(body.assignee_id)}) await _notify( session, ws_id=ctx.ws_id, user_id=body.assignee_id, kind="ticket_assigned", item_id=ticket.id, title=f"转给你一条工单：{ticket.title}", body="请先接收，再开始办理", ) await session.flush() return await _ticket_out(session, ctx.ws_id, ticket, ext) # -- 工单列表 ----------------------------------------------------------- @router.get("/tickets", summary="工单列表：我指派的 / 我发起的") async def list_tickets( ctx: MemberDep, session: SessionDep, role: Annotated[str, Query(description="assigned|requested|all")] = "assigned", status: Annotated[list[str] | None, Query(description="按状态过滤，可重复")] = None, limit: Annotated[int, Query(ge=1, le=100)] = 20, cursor: Annotated[str | None, Query()] = None, ) -> dict: page = Pagination(limit=limit, cursor=cursor) limit, parsed_cursor = page.parsed() stmt: Select = ( select(Item, TicketExt) .join(TicketExt, TicketExt.item_id == Item.id) .where(Item.workspace_id == ctx.ws_id, Item.kind == "ticket", Item.deleted_at.is_(None)) ) if role == "assigned": stmt = stmt.where(TicketExt.assignee_id == ctx.user.id) elif role == "requested": stmt = stmt.where(TicketExt.requester_id == ctx.user.id) elif role != "all": raise AppError(ErrorCode.BAD_REQUEST, "role 只能是 assigned / requested / all") if status: stmt = stmt.where(Item.status.in_(status)) stmt = stmt.order_by(Item.updated_at.desc(), Item.id.desc()).limit(limit + 1) if parsed_cursor is not None: stmt = stmt.where( or_( Item.updated_at < parsed_cursor.created_at, and_(Item.updated_at == parsed_cursor.created_at, Item.id < parsed_cursor.id), ) ) rows = (await session.execute(stmt)).all() has_more = len(rows) > limit rows = rows[:limit] next_cursor = ( encode_cursor(rows[-1][0].updated_at, rows[-1][0].id) if has_more and rows else None ) tickets = await _items_out(session, ctx.ws_id, [item for item, _ in rows]) return ok( { "tickets": [ {**out, "ticket": _ticket_ext_out(ext)} for out, (_, ext) in zip(tickets, rows, strict=True) ], "next_cursor": next_cursor, "has_more": has_more, } ) # -- 事项关系（子任务 / 相关）------------------------------------------ @router.get("/items/{item_id}/links", summary="这条事项的子项与相关项") async def list_links(item_id: uuid.UUID, ctx: MemberDep, session: SessionDep) -> dict: await _load_item(session, ctx.ws_id, item_id, ("task", "ticket")) # parent 边的方向：from 是 to 的子项。所以"我的子项"= 指向我、且类型为 parent 的 from。 children = ( await session.execute( select(ItemLink, Item) .join(Item, Item.id == ItemLink.from_item_id) .where( ItemLink.workspace_id == ctx.ws_id, ItemLink.to_item_id == item_id, ItemLink.link_type == "parent", ) ) ).all() parents = ( await session.execute( select(ItemLink, Item) .join(Item, Item.id == ItemLink.to_item_id) .where( ItemLink.workspace_id == ctx.ws_id, ItemLink.from_item_id == item_id, ItemLink.link_type == "parent", ) ) ).all() related = ( await session.execute( select(ItemLink, Item) .join( Item, or_( and_(ItemLink.from_item_id == item_id, Item.id == ItemLink.to_item_id), and_(ItemLink.to_item_id == item_id, Item.id == ItemLink.from_item_id), ), ) .where( ItemLink.workspace_id == ctx.ws_id, ItemLink.link_type == "related", or_(ItemLink.from_item_id == item_id, ItemLink.to_item_id == item_id), ) ) ).all() async def _pack(rows: list[Any]) -> list[dict[str, Any]]: packed = await _items_out(session, ctx.ws_id, [item for _, item in rows]) return [ {"link_id": str(link.id), "link_type": link.link_type, "item": out} for (link, _), out in zip(rows, packed, strict=True) ] return ok( { "children": await _pack(children), "parents": await _pack(parents), "related": await _pack(related), } ) @router.post("/items/{item_id}/links", summary="把另一条事项挂成子项/相关项") async def create_link( item_id: uuid.UUID, body: LinkIn, ctx: MemberDep, session: SessionDep ) -> dict: if body.link_type not in ("parent", "related"): raise AppError(ErrorCode.BAD_REQUEST, "link_type 只能是 parent / related") if body.to_item_id == item_id: raise AppError(ErrorCode.LINK_CONFLICT, "不能把自己挂到自己下面") await _load_item(session, ctx.ws_id, body.to_item_id, ("task", "ticket")) await _load_item(session, ctx.ws_id, item_id, ("task", "ticket")) existing = ( await session.execute( select(ItemLink).where( ItemLink.from_item_id == body.to_item_id, ItemLink.to_item_id == item_id, ItemLink.link_type == body.link_type, ) ) ).scalar_one_or_none() if existing is not None: raise AppError(ErrorCode.LINK_CONFLICT, "这条关系已经存在") link = ItemLink( workspace_id=ctx.ws_id, from_item_id=body.to_item_id, to_item_id=item_id, link_type=body.link_type, ) session.add(link) await session.flush() return ok({"id": str(link.id), "link_type": link.link_type}) @router.delete("/items/{item_id}/links/{link_id}", summary="解除关系") async def delete_link( item_id: uuid.UUID, link_id: uuid.UUID, ctx: MemberDep, session: SessionDep ) -> dict: link = await session.get(ItemLink, link_id) if ( link is None or link.workspace_id != ctx.ws_id or item_id not in ( link.from_item_id, link.to_item_id, ) ): raise AppError(ErrorCode.LINK_NOT_FOUND, "关系不存在") await session.delete(link) return ok({"deleted": True}) # -- 通知 --------------------------------------------------------------- def _notification_out(n: Notification, item_title: str | None = None) -> dict[str, Any]: return { "id": str(n.id), "kind": n.kind, "item_id": str(n.item_id) if n.item_id else None, "item_title": item_title, "title": n.title, "body": n.body, "read_at": n.read_at, "created_at": n.created_at, } @notify_router.get("", summary="我的通知（默认只看未读）") async def list_notifications( ctx: MemberDep, session: SessionDep, unread_only: bool = True, limit: Annotated[int, Query(ge=1, le=100)] = 30, cursor: Annotated[str | None, Query()] = None, ) -> dict: page = Pagination(limit=limit, cursor=cursor) limit, parsed_cursor = page.parsed() stmt = select(Notification).where( Notification.workspace_id == ctx.ws_id, Notification.user_id == ctx.user.id ) if unread_only: stmt = stmt.where(Notification.read_at.is_(None)) stmt = stmt.order_by(Notification.created_at.desc(), Notification.id.desc()).limit(limit + 1) if parsed_cursor is not None: stmt = stmt.where( or_( Notification.created_at < parsed_cursor.created_at, and_( Notification.created_at == parsed_cursor.created_at, Notification.id < parsed_cursor.id, ), ) ) rows = (await session.execute(stmt)).scalars().all() has_more = len(rows) > limit rows = list(rows[:limit]) titles: dict[uuid.UUID, str] = {} item_ids = [n.item_id for n in rows if n.item_id] if item_ids: for item_id, title in ( await session.execute(select(Item.id, Item.title).where(Item.id.in_(item_ids))) ).all(): titles[item_id] = title next_cursor = encode_cursor(rows[-1].created_at, rows[-1].id) if has_more and rows else None return ok( { "notifications": [ _notification_out(n, titles.get(n.item_id) if n.item_id else None) for n in rows ], "next_cursor": next_cursor, "has_more": has_more, } ) @notify_router.get("/unread-count", summary="未读数（导航角标用）") async def unread_count(ctx: MemberDep, session: SessionDep) -> dict: from sqlalchemy import func as sa_func total = ( await session.execute( select(sa_func.count(Notification.id)).where( Notification.workspace_id == ctx.ws_id, Notification.user_id == ctx.user.id, Notification.read_at.is_(None), ) ) ).scalar_one() return ok({"unread": int(total)}) @notify_router.post("/{notification_id}/read", summary="标记已读") async def mark_read(notification_id: uuid.UUID, ctx: MemberDep, session: SessionDep) -> dict: n = await session.get(Notification, notification_id) if n is None or n.workspace_id != ctx.ws_id or n.user_id != ctx.user.id: raise AppError(ErrorCode.NOTIFICATION_NOT_FOUND, "通知不存在") if n.read_at is None: n.read_at = datetime.now(timezone.utc) return ok({"read_at": n.read_at}) @notify_router.post("/read-all", summary="全部标记已读") async def mark_all_read(ctx: MemberDep, session: SessionDep) -> dict: from sqlalchemy import update as sa_update result = await session.execute( sa_update(Notification) .where( Notification.workspace_id == ctx.ws_id, Notification.user_id == ctx.user.id, Notification.read_at.is_(None), ) .values(read_at=datetime.now(timezone.utc)) ) return ok({"marked": result.rowcount}) __all__ = ["notify_router", "router"]