"""Approval-flow registry: CRUD, ownership, and the reads the rest of the platform asks for. Three rules shape this module: * **Ownership writes are events.** Every owner change appends an ``ApprovalFlowEvent`` carrying both sides, so "who was responsible for 报销审批 in May" is answerable after the fact. * **A machine may fill a blank; it may not overwrite a person.** Inference writes only where ``owner_source`` is not one a human stood behind (:data:`constants.HUMAN_OWNER_SOURCES`). * **An owner who has left is not coverage.** Assignment refuses a non-active person, and the coverage read counts only owners who are still here — a departed owner gets its own bucket, because the fix differs (a first owner vs. a handover somebody forgot). """ from __future__ import annotations import logging import re import uuid from collections.abc import Sequence from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import Select, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import aliased, selectinload from app.core.exceptions import ConflictError, NotFoundError, ValidationError from app.domains.approval_registry import constants as const from app.domains.approval_registry.models import ApprovalFlow, ApprovalFlowEvent from app.domains.approval_registry.serializers import ( feishu_admin_drift, feishu_admin_drift_from, ) from app.identity.person_models import IdentityPerson logger = logging.getLogger(__name__) _CODE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{1,127}$") def person_label(person: IdentityPerson | None) -> str | None: """Display name, falling back to the work email. ``display_name`` is nullable, and an ownership trail that reads 「负责人 — → —」 because two people happened to have no display name is exactly the row an incident review needs to be readable. """ if person is None: return None return person.display_name or person.work_email def _now() -> datetime: return datetime.now(UTC) def _naive(dt: datetime) -> datetime: return dt.replace(tzinfo=None) if dt.tzinfo is not None else dt # ════════════════════════════════════════════════════════════════════ # Validation # ════════════════════════════════════════════════════════════════════ def normalize_code(raw: str) -> str: """Upper-case and validate a Feishu approval code. Upper-casing is not cosmetic: Feishu emits upper-case UUIDs, and a hand-typed lower-case copy would slip past the UNIQUE index and create a second row for the same flow — two owners, two half-year cards, and a register that says the company has one more approval than it does. """ code = (raw or "").strip().upper() if not _CODE_RE.match(code): raise ValidationError( "approval_code must be 2-128 characters of letters, digits, " "'.', '_' or '-' (Feishu emits an upper-case UUID)", ) return code def _bounded(value: str | None, limit: int, field: str) -> str | None: if value is None: return None text = value.strip() if not text: return None if len(text) > limit: raise ValidationError(f"{field} exceeds {limit} characters") return text def _required(value: str | None, limit: int, field: str) -> str: text = _bounded(value, limit, field) if not text: raise ValidationError(f"{field} is required") return text def _validate_enum(value: str, allowed: tuple[str, ...], field: str) -> str: if value not in allowed: raise ValidationError(f"{field} must be one of: {', '.join(allowed)}") return value # ════════════════════════════════════════════════════════════════════ # Reads # ════════════════════════════════════════════════════════════════════ def _base_query() -> Select[tuple[ApprovalFlow]]: return ( select(ApprovalFlow) .where(ApprovalFlow.deleted_at.is_(None)) .options(selectinload(ApprovalFlow.owner)) # ``populate_existing`` for the same reason the system registry # needs it: the session runs ``expire_on_commit=False``, so # re-reading a row already in the identity map after a transfer # hands back the PREVIOUS owner while the column is correct. .execution_options(populate_existing=True) ) def _owner_gone_clause(owner_alias: Any) -> Any: """"Nobody is actually answering for it" — no owner, or one who has left. The KPI tile and the ``ownerless`` filter must select exactly the same rows, or clicking the number shows a shorter list.""" return or_( ApprovalFlow.owner_person_id.is_(None), owner_alias.employment_status != "active", owner_alias.deleted_at.is_not(None), ) async def get_flow(db: AsyncSession, flow_id: uuid.UUID) -> ApprovalFlow: row = ( await db.execute(_base_query().where(ApprovalFlow.id == flow_id)) ).scalar_one_or_none() if row is None: raise NotFoundError("ApprovalFlow", str(flow_id)) return row async def get_by_code(db: AsyncSession, approval_code: str) -> ApprovalFlow | None: return ( await db.execute( _base_query().where( ApprovalFlow.approval_code == normalize_code(approval_code), ), ) ).scalar_one_or_none() async def flows_with_feishu_admin_drift( db: AsyncSession, ) -> list[ApprovalFlow]: """Live flows whose owner is not among the console's 流程管理员. Filtered in Python, and that is a decision rather than laziness. The predicate reads inside a JSON array (「owner_person_id 在 feishu_admins[*].person_id 里吗」), which SQL can only express here through ``JSON_SEARCH`` with a per-row needle — a construct this register would be betting on TiDB supporting identically to MySQL for no gain: the whole table is one company's approval definitions (49 at the time of writing, bounded by how many flows a company can maintain by hand) and the coverage summary already loads all of them on every page render. Unpaginated for the same reason: this is a worklist somebody drains by hand in the Feishu console, not a directory. If it ever needs a page, the fix is that IT has a much bigger problem than pagination. """ rows = ( await db.execute( _base_query().where( ApprovalFlow.status.in_(const.LIVE_STATUSES), ).order_by( func.coalesce(ApprovalFlow.category, "").asc(), ApprovalFlow.name.asc(), ), ) ).scalars().all() return [r for r in rows if feishu_admin_drift(r)] async def flows_with_stale_drift_stamp( db: AsyncSession, still_drifting: set[uuid.UUID], ) -> list[ApprovalFlow]: """Rows carrying a drift-notified stamp that no longer drift. The digest job nulls these, and doing so is not tidiness. Leaving the stamp behind would silently suppress the SAME flow breaking again later: the next digest would skip it as 「already told them」 about a finding that had been resolved and re-created in between. A stamp is a statement about a specific disagreement, and when that disagreement is gone so is the statement. Not restricted to live flows, unlike the drift read itself: a flow retired while drifting stops being a finding, and its stamp should go with it rather than waiting for somebody to un-retire it. """ stmt = _base_query().where( ApprovalFlow.feishu_admin_drift_notified_at.is_not(None), ) if still_drifting: stmt = stmt.where(ApprovalFlow.id.not_in(still_drifting)) return list((await db.execute(stmt)).scalars().all()) async def list_flows( db: AsyncSession, *, q: str | None = None, status: str | None = None, category: str | None = None, owner_source: str | None = None, owner_person_id: uuid.UUID | None = None, ownerless: bool | None = None, machine_owner: bool | None = None, unconfirmed: bool | None = None, published: bool | None = None, source: str | None = None, page: int = 1, page_size: int = 20, ) -> tuple[list[ApprovalFlow], int]: stmt = _base_query() if q: needle = f"%{q.strip().lower()}%" stmt = stmt.where( or_( func.lower(ApprovalFlow.name).like(needle), func.lower(ApprovalFlow.approval_code).like(needle), func.lower(func.coalesce(ApprovalFlow.category, "")).like(needle), func.lower(func.coalesce(ApprovalFlow.source_group, "")).like(needle), ), ) if status: stmt = stmt.where(ApprovalFlow.status == status) if category: stmt = stmt.where(ApprovalFlow.category == category) if owner_source: stmt = stmt.where(ApprovalFlow.owner_source == owner_source) if source: stmt = stmt.where(ApprovalFlow.source == source) if published is not None: stmt = stmt.where(ApprovalFlow.published.is_(published)) if owner_person_id is not None: stmt = stmt.where(ApprovalFlow.owner_person_id == owner_person_id) if ownerless is not None: gone = aliased(IdentityPerson) stmt = stmt.outerjoin(gone, gone.id == ApprovalFlow.owner_person_id) clause = _owner_gone_clause(gone) stmt = stmt.where(clause if ownerless else ~clause) if machine_owner is not None: # Backs the 「待本人确认」 KPI, and exists because that tile once # counted ``owner_source == 'inferred'`` and clicked through to # the same single value. The moment a second machine source # (``feishu_admin``) appeared, the number and the list it opens # would have disagreed — a KPI whose filter shows fewer rows # than it advertises reads as a bug in the data. machine = ApprovalFlow.owner_source.in_(const.MACHINE_OWNER_SOURCES) named = ApprovalFlow.owner_person_id.is_not(None) stmt = ( stmt.where(machine, named) if machine_owner else stmt.where(~machine | ~named) ) if unconfirmed is not None: # "Unconfirmed" is not the same as "ownerless": a flow can carry # an inferred name nobody has stood behind yet, and that is the # queue the half-year round exists to drain. cutoff = _naive(_now() - timedelta(days=const.CONFIRMATION_VALID_DAYS)) stale = or_( ApprovalFlow.owner_confirmed_at.is_(None), ApprovalFlow.owner_confirmed_at < cutoff, ) stmt = stmt.where(stale if unconfirmed else ~stale) total = ( await db.execute( select(func.count()).select_from(stmt.order_by(None).subquery()), ) ).scalar_one() rows = ( await db.execute( stmt.order_by( # Live first, then by category so the admin page reads # like the console it mirrors, then by name. ApprovalFlow.status.asc(), func.coalesce(ApprovalFlow.category, "").asc(), ApprovalFlow.name.asc(), ) .offset((page - 1) * page_size) .limit(page_size), ) ).scalars().all() return list(rows), int(total) async def list_published(db: AsyncSession) -> list[ApprovalFlow]: """Every flow the all-staff page shows: published, not retired. No role gate above this — it is the company's own list of its own approvals, and the point of publishing it is that an employee can find out which flow to use and who to ask without opening a ticket. Retired rows are excluded because a list that advertises flows nobody can submit sends people down dead ends. """ return list( ( await db.execute( _base_query() .where( ApprovalFlow.published.is_(True), ApprovalFlow.status.in_(const.LIVE_STATUSES), ) .order_by( func.coalesce(ApprovalFlow.category, "").asc(), ApprovalFlow.name.asc(), ), ) ).scalars().all(), ) async def flows_owned_by( db: AsyncSession, person_id: uuid.UUID, *, include_retired: bool = False, ) -> list[ApprovalFlow]: stmt = _base_query().where(ApprovalFlow.owner_person_id == person_id) if not include_retired: stmt = stmt.where(ApprovalFlow.status.in_(const.LIVE_STATUSES)) return list( ( await db.execute( stmt.order_by( func.coalesce(ApprovalFlow.category, "").asc(), ApprovalFlow.name.asc(), ), ) ).scalars().all(), ) async def list_categories(db: AsyncSession) -> list[str]: rows = ( await db.execute( select(func.distinct(ApprovalFlow.category)) .where( ApprovalFlow.deleted_at.is_(None), ApprovalFlow.category.is_not(None), ) .order_by(ApprovalFlow.category.asc()), ) ).scalars().all() return [r for r in rows if r] async def list_events( db: AsyncSession, flow_id: uuid.UUID, *, limit: int = 100, ) -> list[ApprovalFlowEvent]: return list( ( await db.execute( select(ApprovalFlowEvent) .where(ApprovalFlowEvent.flow_id == flow_id) .order_by(ApprovalFlowEvent.created_at.desc()) .limit(limit), ) ).scalars().all(), ) async def coverage_summary(db: AsyncSession) -> dict[str, Any]: """The numbers the admin page leads with. The denominator is live flows only. A retired flow without an owner is not a governance gap, and counting it would drag the number down every time something is switched off correctly. """ owner = aliased(IdentityPerson) rows = ( await db.execute( select( ApprovalFlow.status, ApprovalFlow.owner_person_id, ApprovalFlow.owner_source, ApprovalFlow.owner_confirmed_at, ApprovalFlow.published, # For the 「飞书后台待同步」 count. One extra column on a # register of ~50 rows, decided by the SAME predicate the # tab's list uses — a tile whose number is computed # differently from the list it opens is the bug this # domain has already shipped once (see ``machine_owner``). ApprovalFlow.feishu_admins, ( (owner.employment_status == "active") & owner.deleted_at.is_(None) ).label("owner_active"), ) .outerjoin(owner, owner.id == ApprovalFlow.owner_person_id) .where(ApprovalFlow.deleted_at.is_(None)), ) ).all() live = [r for r in rows if r.status in const.LIVE_STATUSES] owned = [r for r in live if r.owner_person_id is not None and r.owner_active] departed = [ r for r in live if r.owner_person_id is not None and not r.owner_active ] cutoff = _naive(_now() - timedelta(days=const.CONFIRMATION_VALID_DAYS)) confirmed_fresh = [ r for r in owned if r.owner_confirmed_at is not None and _naive(r.owner_confirmed_at) >= cutoff ] return { "total": len(rows), "live": len(live), "active": len([r for r in live if r.status == const.STATUS_ACTIVE]), "inactive": len([r for r in live if r.status == const.STATUS_INACTIVE]), "retired": len([r for r in rows if r.status == const.STATUS_RETIRED]), "owned": len(owned), "ownerless": len([r for r in live if r.owner_person_id is None]), "owner_departed": len(departed), # Live flows where Helix and the Feishu console name different # people. Not a coverage failure — somebody IS accountable here # — but the console is what grants the rights, so the successor # currently has the job without the access. Only a human in the # console can close it; the register's job is to refuse to let # the disagreement stay invisible. "feishu_admin_drift": len([ r for r in live if feishu_admin_drift_from(r.owner_person_id, r.feishu_admins) ]), # Machine-established owners are counted separately from # confirmed ones on purpose: a register that reports 100% # coverage on the strength of its own guesses has measured its # guessing, not its coverage. ``owner_machine`` spans both # machine sources; ``owner_inferred`` stays as the narrower # 「统计推断出来的」 count, which is a different question from # 「飞书声明的」 and worth being able to ask on its own. "owner_inferred": len( [r for r in owned if r.owner_source == const.OWNER_SOURCE_INFERRED], ), "owner_feishu_admin": len( [ r for r in owned if r.owner_source == const.OWNER_SOURCE_FEISHU_ADMIN ], ), "owner_machine": len( [r for r in owned if r.owner_source in const.MACHINE_OWNER_SOURCES], ), "owner_confirmed_fresh": len(confirmed_fresh), "coverage_pct": ( round(len(owned) * 100 / len(live), 1) if live else 0.0 ), "confirmed_pct": ( round(len(confirmed_fresh) * 100 / len(live), 1) if live else 0.0 ), "published": len([r for r in live if r.published]), } # ════════════════════════════════════════════════════════════════════ # People # ════════════════════════════════════════════════════════════════════ async def load_person_optional( db: AsyncSession, person_id: uuid.UUID, ) -> IdentityPerson | None: """The actor row, or ``None`` when the JWT outlived the person.""" return ( await db.execute( select(IdentityPerson) .where(IdentityPerson.id == person_id) .options(selectinload(IdentityPerson.org_memberships)), ) ).scalar_one_or_none() async def load_person(db: AsyncSession, person_id: uuid.UUID) -> IdentityPerson: """Eager-load org memberships: ``IdentityPerson.department`` is a property over that relationship, so touching it on a lazily-loaded person raises ``MissingGreenlet`` under asyncio.""" person = ( await db.execute( select(IdentityPerson) .where( IdentityPerson.id == person_id, IdentityPerson.deleted_at.is_(None), ) .options(selectinload(IdentityPerson.org_memberships)), ) ).scalar_one_or_none() if person is None: raise NotFoundError("Person", str(person_id)) return person def assert_assignable(person: IdentityPerson) -> IdentityPerson: """Refuse to name a departed employee as a flow owner. A flow owned by someone who has left is worse than an ownerless one: the ownerless read chases the blank, while a departed name satisfies every "has an owner" check and the half-year card is sent into a mailbox nobody reads. """ if person.employment_status != "active": raise ConflictError( f"{person.work_email} is {person.employment_status!r} — an " "approval-flow owner must be an active employee.", code="APPROVAL_FLOW_OWNER_NOT_ACTIVE", ) return person # ════════════════════════════════════════════════════════════════════ # Events # ════════════════════════════════════════════════════════════════════ async def add_event( db: AsyncSession, flow: ApprovalFlow, *, event_type: str, actor: IdentityPerson | None = None, body: str | None = None, meta: dict[str, Any] | None = None, ) -> ApprovalFlowEvent: event = ApprovalFlowEvent( flow_id=flow.id, event_type=_validate_enum(event_type, const.EVENT_TYPES, "event_type"), actor_person_id=actor.id if actor else None, actor_name=person_label(actor), body=body, meta=meta, ) db.add(event) return event # ════════════════════════════════════════════════════════════════════ # Writes # ════════════════════════════════════════════════════════════════════ async def create_flow( db: AsyncSession, *, approval_code: str, name: str, category: str | None = None, description: str | None = None, status: str = const.STATUS_ACTIVE, published: bool = True, owner_person_id: uuid.UUID | None = None, actor: IdentityPerson | None = None, source: str = const.SOURCE_MANUAL, announce: str | None = const.HANDOVER_ASSIGNED, ) -> ApprovalFlow: """Register a flow by hand. This is the escape hatch that keeps the register honest: discovery can only see flows somebody has actually submitted, so a brand-new or never-used approval is invisible to it. Typing it in is a first-class path, not a workaround. """ code = normalize_code(approval_code) existing = ( await db.execute( select(ApprovalFlow).where(ApprovalFlow.approval_code == code), ) ).scalar_one_or_none() if existing is not None: if existing.deleted_at is None: raise ConflictError( f"approval flow {code} is already registered", code="APPROVAL_FLOW_EXISTS", ) # A soft-deleted row is revived rather than shadowed by a second # one: the UNIQUE index covers deleted rows too, and its history # is the thing worth keeping. existing.deleted_at = None existing.status = _validate_enum(status, const.STATUSES, "status") existing.name = _required(name, const.MAX_NAME_CHARS, "name") await add_event( db, existing, event_type=const.EVENT_CREATED, actor=actor, body="重新登记（此前已归档）", ) return existing owner = None if owner_person_id is not None: owner = assert_assignable(await load_person(db, owner_person_id)) flow = ApprovalFlow( approval_code=code, name=_required(name, const.MAX_NAME_CHARS, "name"), category=_bounded(category, const.MAX_CATEGORY_CHARS, "category"), description=_bounded(description, const.MAX_DESCRIPTION_CHARS, "description"), status=_validate_enum(status, const.STATUSES, "status"), published=bool(published), source=_validate_enum(source, const.SOURCES, "source"), owner_person_id=owner.id if owner else None, owner_source=( const.OWNER_SOURCE_MANUAL if owner else const.OWNER_SOURCE_UNASSIGNED ), owner_department=(owner.department if owner else None), owner_assigned_at=_naive(_now()) if owner else None, first_seen_at=_naive(_now()), created_by=actor.id if actor else None, ) db.add(flow) await db.flush() await add_event( db, flow, event_type=const.EVENT_CREATED, actor=actor, body=f"登记流程「{flow.name}」", meta={"approval_code": code, "source": flow.source}, ) if owner is not None: await add_event( db, flow, event_type=const.EVENT_OWNER_ASSIGNED, actor=actor, body=f"负责人指派为 {person_label(owner)}", meta={"to_person_id": str(owner.id), "to_name": person_label(owner)}, ) # Registering a flow AGAINST somebody is the same act as # transferring one TO them, and this path had the same silence: # IT fills the form in and the named owner is never told. Same # default as ``set_owner``. ``import_flows`` never reaches this # branch — it creates ownerless and assigns through # ``set_owner`` — so the only opt-out is the request flow, which # sends a better message of its own. if announce is not None: from app.domains.approval_registry import notify try: await notify.notify_owner_assigned( db, [flow], new_owner_id=owner.id, previous_owner_id=None, reason=announce, actor=actor, ) except Exception: # noqa: BLE001 logger.warning( "approval_registry: owner-assigned push failed for " "new flow %s", flow.id, exc_info=True, ) return flow async def set_description( db: AsyncSession, flow: ApprovalFlow, *, description: str | None, actor: IdentityPerson | None = None, ) -> ApprovalFlow: """Write the flow's published 用途说明. Returns the flow. Split out of :func:`update_flow` so the owner's self-service path cannot reach name / status / published: those are IT's, and an endpoint that happened to accept them because it shared a body would be a privilege escalation nobody wrote down. All 49 flows in this tenant had an empty description on the day this shipped — the register could say who to ask and not what the thing was for, which is half of 「该走哪个流程」 missing. The owner is the person who knows. """ clean = _bounded(description, const.MAX_DESCRIPTION_CHARS, "description") if clean == flow.description: return flow before, flow.description = flow.description, clean await add_event( db, flow, event_type=const.EVENT_UPDATED, actor=actor, body=("补充了用途说明" if not before else ("清空了用途说明" if not clean else "修改了用途说明")), meta={"field": "description", "before": before, "after": clean}, ) return flow async def update_flow( db: AsyncSession, flow: ApprovalFlow, *, name: str | None = None, category: str | None = None, description: str | None = None, status: str | None = None, published: bool | None = None, clear_category: bool = False, actor: IdentityPerson | None = None, ) -> ApprovalFlow: """Edit the descriptive columns. **Cannot touch ownership** — that goes through :func:`set_owner` so there is no path that changes accountability without leaving an event behind.""" changes: dict[str, Any] = {} if name is not None: new_name = _required(name, const.MAX_NAME_CHARS, "name") if new_name != flow.name: changes["name"] = (flow.name, new_name) flow.name = new_name if clear_category: if flow.category is not None: changes["category"] = (flow.category, None) flow.category = None elif category is not None: new_cat = _bounded(category, const.MAX_CATEGORY_CHARS, "category") if new_cat != flow.category: changes["category"] = (flow.category, new_cat) flow.category = new_cat if description is not None: flow.description = _bounded( description, const.MAX_DESCRIPTION_CHARS, "description", ) if status is not None: new_status = _validate_enum(status, const.STATUSES, "status") if new_status != flow.status: changes["status"] = (flow.status, new_status) flow.status = new_status if published is not None and bool(published) != flow.published: changes["published"] = (flow.published, bool(published)) flow.published = bool(published) if "category" in changes: old, new = changes["category"] await add_event( db, flow, event_type=const.EVENT_CATEGORY_CHANGED, actor=actor, body=f"分类 {old or '—'} → {new or '—'}", ) if "status" in changes: old, new = changes["status"] await add_event( db, flow, event_type=const.EVENT_STATUS_CHANGED, actor=actor, body=f"状态 {old} → {new}", ) if "published" in changes: await add_event( db, flow, event_type=const.EVENT_PUBLISHED_CHANGED, actor=actor, body=("已公示" if flow.published else "已从公示中移除"), ) if "name" in changes: old, new = changes["name"] await add_event( db, flow, event_type=const.EVENT_RENAMED, actor=actor, body=f"名称 {old} → {new}", ) if changes and not ( {"category", "status", "published", "name"} & set(changes) ): await add_event(db, flow, event_type=const.EVENT_UPDATED, actor=actor) return flow async def set_owner( db: AsyncSession, flow: ApprovalFlow, *, owner_person_id: uuid.UUID | None, actor: IdentityPerson | None = None, owner_source: str = const.OWNER_SOURCE_MANUAL, note: str | None = None, confirmed: bool = False, announce: str | None = const.HANDOVER_ASSIGNED, ) -> ApprovalFlow: """Assign, transfer or clear the owner — always as an event. ``confirmed=True`` additionally stamps ``owner_confirmed_at``: the half-year round and an owner's own "yes, still me" go through here, and the stamp is what makes the difference between "a name is in the box" and "a person stood behind it" measurable. ``announce`` names the handover reason pushed to the **new** owner, and defaults to announcing. It lives here rather than at each call site on purpose: wiring the push per-caller was tried first and had already missed one — an owner transferring their flow between rounds goes through ``record_self_answer``, which writes the register directly, and that path stayed silent. Defaulting to announce means a future path gets the card for free and opting out is a deliberate ``announce=None`` with a stated reason. """ _validate_enum(owner_source, const.OWNER_SOURCES, "owner_source") new_owner: IdentityPerson | None = None if owner_person_id is not None: new_owner = assert_assignable(await load_person(db, owner_person_id)) previous_id = flow.owner_person_id previous_name: str | None = None if previous_id is not None: prior = await load_person_optional(db, previous_id) previous_name = person_label(prior) unchanged = previous_id == (new_owner.id if new_owner else None) flow.owner_person_id = new_owner.id if new_owner else None flow.owner_department = new_owner.department if new_owner else None if new_owner is None: flow.owner_source = const.OWNER_SOURCE_UNASSIGNED flow.owner_assigned_at = None flow.owner_confirmed_at = None flow.inference_meta = None else: flow.owner_source = ( const.OWNER_SOURCE_CONFIRMED if confirmed else owner_source ) if not unchanged: flow.owner_assigned_at = _naive(_now()) if confirmed: flow.owner_confirmed_at = _naive(_now()) if owner_source != const.OWNER_SOURCE_INFERRED: # A human decision supersedes the machine's evidence; keeping # the old inference next to it invites someone to read a # stale justification as support for a name it never chose. flow.inference_meta = None if unchanged and new_owner is not None: # Re-confirming the same person is a confirmation event, not a # transfer — a timeline row reading "A → A" is noise that makes # the real handovers harder to find. await add_event( db, flow, event_type=( const.EVENT_OWNER_CONFIRMED if confirmed else const.EVENT_OWNER_ASSIGNED ), actor=actor, body=note or ( f"确认负责人仍为 {person_label(new_owner)}" if confirmed else None ), meta={"to_person_id": str(new_owner.id), "to_name": person_label(new_owner)}, ) return flow if new_owner is None: await add_event( db, flow, event_type=const.EVENT_OWNER_CLEARED, actor=actor, body=note or f"移除负责人（原 {previous_name or '—'}）", meta={ "from_person_id": str(previous_id) if previous_id else None, "from_name": previous_name, }, ) return flow await add_event( db, flow, event_type=( const.EVENT_OWNER_TRANSFERRED if previous_id is not None else const.EVENT_OWNER_ASSIGNED ), actor=actor, body=note or ( f"负责人 {previous_name or '—'} → {person_label(new_owner)}" if previous_id is not None else f"负责人指派为 {person_label(new_owner)}" ), meta={ "from_person_id": str(previous_id) if previous_id else None, "from_name": previous_name, "to_person_id": str(new_owner.id), "to_name": person_label(new_owner), "owner_source": flow.owner_source, }, ) # Tell the person who just became responsible. Deliberately after # every mutation above, so the card describes the row as it now is. # ``notify`` imports this module, so the import is local — the same # seam the coordinator uses. if announce is not None and not unchanged and new_owner is not None: from app.domains.approval_registry import notify try: await notify.notify_owner_assigned( db, [flow], new_owner_id=new_owner.id, previous_owner_id=previous_id, reason=announce, actor=actor, ) except Exception: # noqa: BLE001 # The router's own pattern is 「先 commit 业务，再通知」 — a # push must never undo the thing it is announcing. Living # inside the setter buys 「没有调用点会忘」 and costs that # ordering, so the guard buys it back: the ownership change # stands, the card is lost, and the loss is logged rather # than silent. logger.warning( "approval_registry: owner-assigned push failed for flow %s", flow.id, exc_info=True, ) # And tell IT, at the moment of the transfer, that the Feishu console # now disagrees with this row — the one step nothing here can perform # (user decision, 2026-08-28, alongside the daily digest). # # Gated on ``announce == HANDOVER_TRANSFER`` rather than on "any # ownership change", and that narrowness is the whole reason it can # coexist with the digest without becoming the storm the digest was # built to avoid. HANDOVER_TRANSFER is passed from exactly two call # sites, both an employee pressing the button themselves; the # offboarding coordinator and the request-registration paths pass # ``announce=None``, so a batch hand-over that moves nine flows in # one transaction still surfaces as one digest line. # # Gated on the gap actually existing, too: a transfer to somebody the # console ALREADY names needs no console edit, and an alert with no # action behind it is the fastest way to get this channel muted. if ( announce == const.HANDOVER_TRANSFER and not unchanged and new_owner is not None # ``actor`` is optional on this setter, but 「谁转交的」 is the # first line of the card. Both HANDOVER_TRANSFER call sites pass # a real person, so this is unreachable; if a future one does # not, the row still lands in the 09:00 digest rather than # producing a card addressed by nobody. and actor is not None and feishu_admin_drift(flow) ): from app.domains.approval_registry import notify try: reached = await notify.notify_transfer_console_gap( db, flow, actor=actor, successor=new_owner, previous_owner_name=previous_name, ) if reached: # Claim the digest's next mention of this flow — IT has # just been told, and hearing it again at 09:00 tomorrow # teaches them the digest repeats itself. Stamped ONLY on # a delivered push: a failed alert must leave this NULL # so the digest is a safety net rather than a duplicate. flow.feishu_admin_drift_notified_at = _naive(_now()) except Exception: # noqa: BLE001 — same reason as the push above logger.warning( "approval_registry: transfer console-gap alert failed for " "flow %s", flow.id, exc_info=True, ) return flow async def confirm_owner( db: AsyncSession, flow: ApprovalFlow, *, actor: IdentityPerson, note: str | None = None, ) -> ApprovalFlow: """The owner says "yes, still me". Idempotent by nature — a second click just re-stamps the date.""" if flow.owner_person_id != actor.id: raise ConflictError( "only the current owner can confirm this flow", code="APPROVAL_FLOW_NOT_OWNER", ) return await set_owner( db, flow, owner_person_id=actor.id, actor=actor, owner_source=const.OWNER_SOURCE_CONFIRMED, confirmed=True, note=note, # 「还是我」 is not a handover. The ``unchanged`` guard inside # already covers it; saying so here keeps the reader from # wondering whether confirming mails yourself a card. announce=None, ) async def dispute_owner( db: AsyncSession, flow: ApprovalFlow, *, actor: IdentityPerson, reason: str | None = None, ) -> ApprovalFlowEvent: """"This is not mine" — recorded on the flow's own trail, next to the ownership history it disputes. It does NOT clear the owner: leaving the register blank on one person's say-so would let a contested flow vanish from the coverage read instead of surfacing for IT to re-point.""" if flow.owner_person_id != actor.id: raise ConflictError( "only the named owner can dispute this flow's ownership", code="APPROVAL_FLOW_NOT_OWNER", ) return await add_event( db, flow, event_type=const.EVENT_OWNER_DISPUTED, actor=actor, body=_bounded(reason, const.MAX_DESCRIPTION_CHARS, "reason") or "负责人表示此流程不归自己负责", meta={"person_id": str(actor.id)}, ) async def retire_flow( db: AsyncSession, flow: ApprovalFlow, *, actor: IdentityPerson | None = None, reason: str | None = None, ) -> ApprovalFlow: """Archive a flow. Sets the status AND soft-deletes, and both exits are excluded from the coverage denominator — the soft-delete filter alone would hide a missing status check.""" flow.status = const.STATUS_RETIRED flow.deleted_at = _naive(_now()) await add_event( db, flow, event_type=const.EVENT_RETIRED, actor=actor, body=_bounded(reason, const.MAX_DESCRIPTION_CHARS, "reason") or "已归档", ) return flow # ════════════════════════════════════════════════════════════════════ # Sweep-facing writes (discovery / definition sync / inference) # ════════════════════════════════════════════════════════════════════ async def upsert_discovered( db: AsyncSession, *, approval_code: str, name: str, source_group: str | None, observed_instances: int, last_instance_at: datetime | None, ) -> tuple[ApprovalFlow, bool]: """Record what the discovery sweep saw. Returns ``(flow, created)``. Upstream fields (``name``, ``source_group``) are refreshed; the Helix-side ones (``category``, ``description``, ownership) are never touched here — a nightly sweep that overwrote a human's classification would make the register unusable within a week. """ code = normalize_code(approval_code) row = ( await db.execute( select(ApprovalFlow).where(ApprovalFlow.approval_code == code), ) ).scalar_one_or_none() now = _naive(_now()) clean_name = _required(name, const.MAX_NAME_CHARS, "name") group = _bounded(source_group, const.MAX_CATEGORY_CHARS, "source_group") if row is None: row = ApprovalFlow( approval_code=code, name=clean_name, source_group=group, # Seeded from upstream, then owned by IT. The two columns # disagree in this tenant, so the seed is a starting point, # not a claim. category=group, source=const.SOURCE_DISCOVERED, first_seen_at=now, last_discovered_at=now, observed_instances=observed_instances, last_instance_at=last_instance_at, ) db.add(row) await db.flush() await add_event( db, row, event_type=const.EVENT_DISCOVERED, body=f"自动发现流程「{clean_name}」", meta={ "approval_code": code, "source_group": group, "observed_instances": observed_instances, }, ) return row, True if row.deleted_at is not None: # Seen again after being archived. Say so on the trail instead # of quietly reviving it: somebody archived it for a reason and # the flow is evidently still in use. await add_event( db, row, event_type=const.EVENT_DISCOVERED, body="已归档的流程再次出现在飞书审批实例中", meta={"observed_instances": observed_instances}, ) elif clean_name != row.name: await add_event( db, row, event_type=const.EVENT_RENAMED, body=f"飞书侧名称 {row.name} → {clean_name}", ) row.name = clean_name if group and group != row.source_group: row.source_group = group row.last_discovered_at = now row.observed_instances = observed_instances if last_instance_at is not None and ( row.last_instance_at is None or _naive(last_instance_at) > _naive(row.last_instance_at) ): row.last_instance_at = last_instance_at if row.first_seen_at is None: row.first_seen_at = now return row, False async def apply_definition( db: AsyncSession, flow: ApprovalFlow, *, name: str | None, feishu_status: str | None, error: str | None = None, ) -> ApprovalFlow: """Fold one ``GET /approval/v4/approvals/{code}`` result into the row: the upstream name and status, and nothing else. The definition also carries the form and the node list; we deliberately do not store either.""" now = _naive(_now()) if error is not None: flow.last_sync_error = error[:512] return flow flow.last_sync_error = None flow.last_synced_at = now if name: clean = _bounded(name, const.MAX_NAME_CHARS, "name") if clean and clean != flow.name: await add_event( db, flow, event_type=const.EVENT_RENAMED, body=f"飞书侧名称 {flow.name} → {clean}", ) flow.name = clean if feishu_status: flow.feishu_status = feishu_status[:32] mapped = const.FEISHU_STATUS_MAP.get(feishu_status.upper()) # An unmapped upstream status leaves ours alone: guessing # "active" for a value Feishu invented later would silently # resurrect a stopped flow on the all-staff page. if mapped and mapped != flow.status: await add_event( db, flow, event_type=const.EVENT_STATUS_CHANGED, body=f"飞书侧状态 {flow.status} → {mapped}", meta={"feishu_status": feishu_status}, ) flow.status = mapped return flow def _admin_key(admins: list[dict[str, Any]] | None) -> list[str]: """Comparable identity of an admin list — the user_ids, sorted. Names and employment flags churn (somebody changes their display name, somebody resigns); the *membership* is what an event should fire on. Comparing whole dicts would write 「管理员变更」 to the trail every time a resignation flag flipped. """ return sorted( str(a.get("user_id")) for a in (admins or []) if a.get("user_id") ) async def apply_feishu_admins( db: AsyncSession, flow: ApprovalFlow, *, admins: list[dict[str, Any]], ) -> bool: """Store the definition's 流程管理员 snapshot. Returns whether the membership changed. Always stamps ``feishu_admins_synced_at``, including for an empty list: 「we asked and Feishu named nobody」 is a different, and useful, fact from 「we never asked」 — and telling them apart is the whole reason the column exists. """ trimmed = admins[: const.MAX_FEISHU_ADMINS] before, after = _admin_key(flow.feishu_admins), _admin_key(trimmed) flow.feishu_admins = trimmed flow.feishu_admins_synced_at = _naive(_now()) if before == after: return False await add_event( db, flow, event_type=const.EVENT_FEISHU_ADMINS_CHANGED, body=( "飞书流程管理员：{}".format( "、".join(a.get("name") or a.get("email") or "?" for a in trimmed) or "无", ) ), meta={"before": before, "after": after}, ) return True async def apply_feishu_admin_owner( db: AsyncSession, flow: ApprovalFlow, *, person: IdentityPerson, ) -> bool: """Adopt the definition's sole 流程管理员 as the flow's owner. Returns whether anything changed. Same refusal as inference for a human-established owner, and the same refusal for a leaver — but unlike inference it MAY overwrite an ``inferred`` owner, because a declaration in the Feishu console outranks a count of who happens to click Approve. The reverse is blocked in :func:`apply_inferred_owner`, so the two sweeps cannot take turns flipping the same row. """ if flow.owner_source in const.HUMAN_OWNER_SOURCES: return False if person.employment_status != "active" or person.deleted_at is not None: return False if ( flow.owner_person_id == person.id and flow.owner_source == const.OWNER_SOURCE_FEISHU_ADMIN ): return False previous_id = flow.owner_person_id previous_source = flow.owner_source flow.owner_person_id = person.id flow.owner_source = const.OWNER_SOURCE_FEISHU_ADMIN flow.owner_department = person.department flow.owner_assigned_at = _naive(_now()) # Not carried over from an inferred predecessor: nobody has stood # behind THIS name yet, and 待确认 is the honest render. flow.owner_confirmed_at = None await add_event( db, flow, event_type=const.EVENT_OWNER_FROM_FEISHU_ADMIN, body=( f"飞书审批定义的流程管理员只有 {person_label(person)} 一人，" f"台账据此认领（待本人确认）" ), meta={ "from_person_id": str(previous_id) if previous_id else None, "from_owner_source": previous_source, "to_person_id": str(person.id), "to_name": person_label(person), }, ) return True async def apply_inferred_owner( db: AsyncSession, flow: ApprovalFlow, *, person: IdentityPerson, evidence: dict[str, Any], ) -> bool: """Write a machine-derived owner. Returns whether anything changed. Refuses to touch a row whose owner a human established (:data:`constants.HUMAN_OWNER_SOURCES`) — including one an employee confirmed months ago. The sweep exists to fill blanks, not to re-litigate decisions. Also refuses to overwrite a Feishu-declared 流程管理员: task counts are the weaker evidence, and two machine sweeps that each overwrote the other would re-point the row on alternating nights. """ if flow.owner_source not in const.INFERENCE_OVERWRITABLE_SOURCES: return False if person.employment_status != "active" or person.deleted_at is not None: # Inferring a leaver is worse than inferring nobody: it fills # the blank the coverage read would otherwise chase. return False if flow.owner_person_id == person.id: flow.inference_meta = evidence return False previous_id = flow.owner_person_id flow.owner_person_id = person.id flow.owner_source = const.OWNER_SOURCE_INFERRED flow.owner_department = person.department flow.owner_assigned_at = _naive(_now()) flow.owner_confirmed_at = None flow.inference_meta = evidence await add_event( db, flow, event_type=const.EVENT_OWNER_INFERRED, body=( f"根据实际审批记录推断负责人为 {person_label(person)}" f"（占比 {evidence.get('share_pct', '—')}%，" f"{evidence.get('task_count', '—')} 次决策）" ), meta={ "from_person_id": str(previous_id) if previous_id else None, "to_person_id": str(person.id), "to_name": person_label(person), "evidence": evidence, }, ) return True # ════════════════════════════════════════════════════════════════════ # Bulk import # ════════════════════════════════════════════════════════════════════ async def import_flows( db: AsyncSession, rows: list[dict[str, Any]], *, actor: IdentityPerson | None = None, ) -> dict[str, Any]: """Bulk-register flows from a pasted table. Per-row isolation: one bad code does not lose the other 200 rows, and every rejection is returned with its reason so the operator can fix that line instead of guessing which one failed. """ if len(rows) > const.MAX_IMPORT_ROWS: raise ValidationError( f"at most {const.MAX_IMPORT_ROWS} rows per import", ) created = 0 updated = 0 errors: list[dict[str, Any]] = [] for index, raw in enumerate(rows): try: code = normalize_code(str(raw.get("approval_code") or "")) name = _required( str(raw.get("name") or ""), const.MAX_NAME_CHARS, "name", ) category = _bounded( raw.get("category"), const.MAX_CATEGORY_CHARS, "category", ) owner_email = (raw.get("owner_email") or "").strip().lower() or None owner: IdentityPerson | None = None if owner_email: owner = ( await db.execute( select(IdentityPerson).where( func.lower(IdentityPerson.work_email) == owner_email, IdentityPerson.deleted_at.is_(None), ) .options(selectinload(IdentityPerson.org_memberships)), ) ).scalar_one_or_none() if owner is None: raise ValidationError(f"no person with work email {owner_email}") assert_assignable(owner) existing = await get_by_code(db, code) if existing is None: flow = await create_flow( db, approval_code=code, name=name, category=category, description=raw.get("description"), actor=actor, source=const.SOURCE_IMPORTED, ) created += 1 else: flow = existing await update_flow( db, flow, name=name, category=category, description=raw.get("description"), actor=actor, ) updated += 1 if owner is not None and flow.owner_person_id != owner.id: await set_owner( db, flow, owner_person_id=owner.id, actor=actor, owner_source=const.OWNER_SOURCE_IMPORTED, note="批量导入指派", # Bulk. The first paste-import IS how this register # was seeded — announcing here would card most of # the company in one click, which is the storm the # opening cron's window guard exists to prevent. announce=None, ) await db.flush() except (ValidationError, ConflictError, NotFoundError) as exc: errors.append({ "row": index + 1, "approval_code": str(raw.get("approval_code") or ""), "error": str(exc), }) return { "created": created, "updated": updated, "failed": len(errors), "errors": errors, } async def flows_by_ids( db: AsyncSession, flow_ids: list[uuid.UUID], ) -> list[ApprovalFlow]: """Bulk fetch for the round views — one query rather than one per item, and it deliberately includes archived rows: a round item about a flow archived mid-round still has to render its name.""" if not flow_ids: return [] return list( ( await db.execute( select(ApprovalFlow).where(ApprovalFlow.id.in_(set(flow_ids))), ) ).scalars().all(), ) async def people_by_ids( db: AsyncSession, person_ids: Sequence[uuid.UUID | None], ) -> dict[uuid.UUID | None, IdentityPerson]: """``{person_id: person}`` for the round views. Keyed with an optional id so a caller can look up a nullable column (``transferred_to_person_id`` is null on most items) without narrowing at every call site — a missing key returns ``None``, which is exactly what the serializer renders as "—". """ wanted = {p for p in person_ids if p is not None} if not wanted: return {} rows = ( await db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(wanted)), ) ).scalars().all() return {p.id: p for p in rows} async def flows_by_codes( db: AsyncSession, approval_codes: Sequence[str], ) -> list[ApprovalFlow]: """Bulk lookup by approval code for the nightly sweep. Codes are normalised on the way in for the same reason writes are: a lower-case copy would silently miss its row and the sweep would report the flow as unreadable rather than updating it. Soft-deleted rows are excluded, matching :func:`get_by_code`. """ wanted: set[str] = set() for raw in approval_codes: if not raw: continue try: wanted.add(normalize_code(raw)) except ValidationError: # One code the vendor returned in a shape we do not accept # must not take out a bulk read the whole sweep depends on. continue if not wanted: return [] return list( ( await db.execute( _base_query().where(ApprovalFlow.approval_code.in_(wanted)), ) ).scalars().all(), )