import json import re import unicodedata from datetime import UTC, date, datetime, timedelta from decimal import Decimal from typing import Any from uuid import UUID from zoneinfo import ZoneInfo from fastapi import UploadFile # Aliased: the Helix ``ValidationError`` (an ``AppException`` subclass) is # imported below under the bare name. Before this alias the second import # silently shadowed pydantic's, so ``except (AppException, ValidationError)`` # — written to also catch a model-construction failure — caught only # Helix errors. Pydantic's happens to subclass ``ValueError``, which # masked the bug wherever ``ValueError`` was in the same tuple; the bulk # intake claim handler (no ``ValueError``) was the one place it wasn't. from pydantic import ValidationError as PydanticValidationError from sqlalchemy import and_, asc, desc, func, or_, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import aliased, selectinload from app.approval.cancel_desync import notify_feishu_cancel_desync from app.approval.feishu_client import FeishuApprovalClient from app.approval.models import ApprovalRequest from app.auth.models import AuthUser from app.core.exceptions import AppException, NotFoundError, ValidationError from app.core.person_scope import SubordinateScope, subordinate_person_ids from app.core.safe_logging import get_safe_logger from app.core.txn import best_effort from app.domains.asset.bulk_import import read_bulk_rows from app.domains.asset.location_resolver import ManagedLocationResolver from app.domains.asset.models import ( ASSET_HOLDER_KIND_OFFICE, ASSET_HOLDER_KIND_PERSONAL, ASSET_IN_USE_STATUSES, ASSET_STATUS_HELD_BY_LEAVER, ASSET_STATUS_HELD_BY_OFFICE, AssetCategoryConfig, AssetEmployeeRequest, AssetHistory, AssetItem, AssetLegalEntity, AssetLocation, AssetOfficeAddress, AssetOperation, AssetProcurementOrder, AssetSupplier, ) from app.domains.asset.schemas import ( AssetAssignRequest, AssetBorrowRequest, AssetBorrowReturnRequest, AssetBulkRecycleToOfficeRequest, AssetCategoryConfigCreateRequest, AssetCategoryConfigUpdateRequest, AssetChangeAssigneeRequest, AssetClaimRequest, AssetCreateRequest, AssetDisposeRequest, AssetIntakeRequest, AssetLocationCreateRequest, AssetLocationUpdateRequest, AssetOperationResponse, AssetRepairCompleteRequest, AssetRepairRequest, AssetReturnRequest, AssetStatsResponse, AssetTransferRequest, AssetUpdateRequest, BulkAssetMutationResponse, BulkAssetRowResult, BulkBorrowRequest, BulkClaimRequest, EmployeeRequestCreateRequest, ProcurementOrderCreateRequest, ProcurementOrderUpdateRequest, SupplementUploadFile, sanitize_disposal_photo_keys, ) from app.integrations.compat import get_feishu_config logger = get_safe_logger(__name__) def _esc(s: str) -> str: """Escape LIKE metacharacters to prevent wildcard injection.""" return s.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") def _canonical_purchase_entity_name(value: str) -> str: """Return the safe comparison key for a legal-entity display name. ``asset_items.purchase_entity`` predates the legal-entity FK-shaped picker and is still a text column. Treat harmless formatting copied from invoices (full-width characters, invisible whitespace, and punctuation in English company suffixes) as equivalent, but keep the comparison exact in every other respect. """ normalized = unicodedata.normalize("NFKC", value) normalized = normalized.replace("\u200b", "").replace("\ufeff", "") normalized = re.sub(r"\s+", " ", normalized).strip() # ``LTD`` and ``LTD.`` are the same legal-suffix spelling. Do not remove # arbitrary characters: a broader fuzzy match could select the wrong bank # details for a similarly named entity. return normalized.replace(".", "").casefold() def _asset_duplicate_unique_column(exc: IntegrityError) -> str | None: """Return which ``asset_items`` UNIQUE column a MySQL/TiDB 1062 error hit (``'asset_tag'`` or ``'serial_number'``), or ``None`` if it isn't a 1062 on one of those keys. The 1062 message names the *index*, not the column — and the name differs by environment: migrations name it ``uk_asset_items_serial_number`` (prod/dev) while tests' ``create_all`` names it ``serial_number``. So we pull the key name out of ``for key '...'`` and look for the column token inside it, instead of assuming the index is named after the column. The old ``"asset_items.serial_number" in err`` check matched only the test name, so a real soft-deleted-serial collision in prod fell through to an unhandled 500 instead of the friendly 409. """ orig = getattr(exc, "orig", None) args = getattr(orig, "args", None) if not args or args[0] != 1062: return None match = re.search(r"for key '([^']+)'", str(orig)) key = (match.group(1) if match else "").lower() if "asset_tag" in key: return "asset_tag" if "serial_number" in key: return "serial_number" return None def _generate_operation_number() -> str: """Generate a unique operation number like OP-20260309153000123456-XXXX.""" now = datetime.now(UTC) import secrets suffix = secrets.token_hex(2).upper() return f"OP-{now.strftime('%Y%m%d%H%M%S%f')}-{suffix}" def _generate_order_number() -> str: """Generate a unique procurement order number like PO-20260309-XXXX.""" now = datetime.now(UTC) import secrets suffix = secrets.token_hex(2).upper() return f"PO-{now.strftime('%Y%m%d')}-{suffix}" def _generate_request_number() -> str: """Generate a unique request number like REQ-20260309-XXXX.""" now = datetime.now(UTC) import secrets suffix = secrets.token_hex(2).upper() return f"REQ-{now.strftime('%Y%m%d')}-{suffix}" def _parse_optional_date(value: Any) -> date | None: if value in (None, ""): return None if isinstance(value, date) and not isinstance(value, datetime): return value if isinstance(value, datetime): return value.date() return date.fromisoformat(str(value).strip()) def _normalize_bulk_row(row: dict[str, Any]) -> dict[str, Any]: converters: dict[str, Any] = { "purchase_date": _parse_optional_date, "warranty_until": _parse_optional_date, "expected_return_date": _parse_optional_date, "purchase_cost": lambda v: None if v in (None, "") else Decimal(str(v).strip()), "repair_total_cost": lambda v: None if v in (None, "") else Decimal(str(v).strip()), "usage_limit_months": lambda v: None if v in (None, "") else int(v), "repair_count": lambda v: None if v in (None, "") else int(v), "epandian_id": lambda v: None if v in (None, "") else int(v), "admin_id": lambda v: None if v in (None, "") else UUID(str(v).strip()), "assigned_to": lambda v: None if v in (None, "") else UUID(str(v).strip()), } normalized: dict[str, Any] = {} for key, value in row.items(): cleaned_key = str(key).strip() if not cleaned_key: continue if isinstance(value, str): value = value.strip() if value in ("", None): continue converter = converters.get(cleaned_key) normalized[cleaned_key] = converter(value) if converter else value return normalized async def serialize_operations( db: AsyncSession, ops: list[AssetOperation], *, admin_view: bool = True, ) -> list[AssetOperationResponse]: """Turn a list of ``AssetOperation`` rows into response objects with every denormalised display field populated — ``operator_name/email``, ``from_user_name/email``, ``to_user_name/email``, ``asset_tag``, ``asset_name`` — the data the 操作记录 table needs to render readable rows. Why this helper exists: ``AssetOperation`` only stores FK columns (operator_id / from_user_id / to_user_id / asset_id), and the ``AssetOperationResponse`` schema has the denormalised label fields but no ORM relationship to feed them. So ``AssetOperationResponse.model_validate(op)`` would leave every label as ``None`` and every row in ``/admin/assets/operations`` rendered "-" across 操作人 / 资产编码 / 资产名称 — despite the FKs being populated. This helper does two batch queries (persons + assets) and fills the labels in one pass, keeping list endpoints at a fixed cost rather than O(N) extra round-trips. ``admin_view`` (default True for back-compat) mirrors the gating semantics of ``AssetService.enrich_operations_with_users``: when False, the ``detail`` dict is sanitised to remove the ``location_raw`` key (an HRIS-derived raw value the resolver canonicalised at write time, admin-only audit signal). The ``GET /assets/operations`` list endpoint passes the caller's role state through this gate so a non-admin employee querying their own operation history can't read the operator-submitted raw location values. Round-5 review high. """ from app.identity.person_models import IdentityPerson person_ids: set[UUID] = set() asset_ids: set[UUID] = set() for op in ops: if op.operator_id: person_ids.add(op.operator_id) if op.from_user_id: person_ids.add(op.from_user_id) if op.to_user_id: person_ids.add(op.to_user_id) if op.asset_id: asset_ids.add(op.asset_id) pmap: dict[UUID, IdentityPerson] = {} if person_ids: rows = ( await db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(person_ids)) ) ).scalars().all() pmap = {row.id: row for row in rows} amap: dict[UUID, AssetItem] = {} if asset_ids: rows = ( await db.execute( select(AssetItem).where(AssetItem.id.in_(asset_ids)) ) ).scalars().all() amap = {row.id: row for row in rows} out: list[AssetOperationResponse] = [] for op in ops: resp = AssetOperationResponse.model_validate(op) # Mutate the response in-place instead of rebuilding — keeps the # schema's other defaults (notes, detail, confirmed_* etc) intact. if op.operator_id and (p := pmap.get(op.operator_id)): resp.operator_name = p.display_name resp.operator_email = p.work_email resp.operator_employee_no = p.employee_no resp.operator_employment_status = p.employment_status if op.from_user_id and (p := pmap.get(op.from_user_id)): resp.from_user_name = p.display_name resp.from_user_email = p.work_email resp.from_user_employee_no = p.employee_no resp.from_user_employment_status = p.employment_status if op.to_user_id and (p := pmap.get(op.to_user_id)): resp.to_user_name = p.display_name resp.to_user_email = p.work_email resp.to_user_employee_no = p.employee_no resp.to_user_employment_status = p.employment_status if op.asset_id and (a := amap.get(op.asset_id)): resp.asset_tag = a.asset_tag resp.asset_name = a.asset_name # Non-admin viewers (employee self-scope on # ``GET /assets/operations?to_user_id=``, manager scope # on ``GET /assets/operations?to_user_id=``) must NOT # see the operator-submitted raw location string. Mirror the # ``enrich_operations_with_users`` redaction so both list and # detail surfaces share one PII gate. Operates on the # response object's detail dict — the ORM row's detail # JSON is not mutated. if not admin_view and resp.detail: resp.detail = { k: v for k, v in resp.detail.items() if k != "location_raw" } out.append(resp) return out async def serialize_operation( db: AsyncSession, op: AssetOperation, *, admin_view: bool = True, ) -> AssetOperationResponse: """Single-row variant of :func:`serialize_operations`. All single-op endpoints (create/confirm/intake/...) use this so the callback payload carries the same operator_name / email values the list endpoint now surfaces — otherwise the post-action UI would flicker back to "-" between the mutation response and the list refetch. ``admin_view`` propagates to ``serialize_operations``; default True for back-compat with single-op endpoints, which are all admin-gated. """ results = await serialize_operations(db, [op], admin_view=admin_view) return results[0] class AssetService: # Outcome of the most recent ``cancel_employee_request()`` call's # best-effort Feishu side-cancel. Routers / MCP tools read it after # the call to surface ``feishu_cancel_synced`` to the requester. last_feishu_cancel_synced: bool = True def __init__(self, db: AsyncSession): self.db = db # Lazy-loaded managed-location resolver. Cached on the service # instance (one ``AssetService`` per request) so every write # boundary in a single request — including N iterations of a # bulk operation — shares the SAME master snapshot. Without # this, ``bulk_claim_assets`` would call ``ManagedLocationResolver.load()`` # once per asset, which is both wasteful AND a documented-vs- # actual idempotency violation (PLATFORM.md states the # resolver master is "per-request"; without instance caching # an admin editing ``asset_locations`` mid-bulk would resolve # some rows against the old master and others against the # new — Round-3 review blocker). self._location_resolver: ManagedLocationResolver | None = None async def _get_location_resolver(self) -> ManagedLocationResolver: """Return the per-request location resolver, loading once.""" if self._location_resolver is None: self._location_resolver = ( await ManagedLocationResolver.load(self.db) ) return self._location_resolver async def _resolve_canonical_purchase_entity_name( self, raw_name: str, ) -> str: """Resolve a supplied legal-entity label to its stored canonical name. New asset writes must not preserve a free-text variant that later fails the approval payment-information lookup. We accept only a unique active legal-entity match; unknown or ambiguous labels fail loudly instead of silently attaching the wrong payment details. """ key = _canonical_purchase_entity_name(raw_name) if not key: raise AppException( code="PURCHASE_ENTITY_NOT_REGISTERED", message="Purchase entity must match a registered legal entity", status_code=400, ) rows = ( await self.db.execute( select(AssetLegalEntity).where(AssetLegalEntity.is_active.is_(True)) ) ).scalars().all() matches = [ row for row in rows if _canonical_purchase_entity_name(row.name) == key ] if not matches: raise AppException( code="PURCHASE_ENTITY_NOT_REGISTERED", message="Purchase entity must match a registered legal entity", status_code=400, ) if len(matches) > 1: raise AppException( code="PURCHASE_ENTITY_AMBIGUOUS", message="Purchase entity matches more than one registered legal entity", status_code=400, ) return matches[0].name # ------------------------------------------------------------------ # Existing CRUD operations (backward compatible) # ------------------------------------------------------------------ async def create_asset( self, data: AssetCreateRequest, operator_id: UUID, ) -> AssetItem: """Create a new asset.""" if data.purchase_entity is not None: data.purchase_entity = await self._resolve_canonical_purchase_entity_name( data.purchase_entity ) # Round 2b audit fix-up 2026-05-04: filter ``deleted_at IS NULL`` # on uniqueness probes so a soft-deleted asset's tag/serial # can be reused for a new physical asset (the prior model # uniqueness wasn't released by soft-delete; admin had to # rename to recycle). The DB still has a UNIQUE constraint # on ``asset_tag`` / ``serial_number``; if a soft-deleted row # is found here we surface a friendlier "tag exists on a # soft-deleted asset; restore or rename" error rather than the # raw 1062 IntegrityError that the old service-level check # would let through. existing = await self.db.execute( select(AssetItem).where( AssetItem.asset_tag == data.asset_tag, AssetItem.deleted_at.is_(None), ) ) if existing.scalar_one_or_none(): raise AppException( code="ASSET_TAG_EXISTS", message=f"Asset tag {data.asset_tag} already exists", status_code=400, ) # Check serial number uniqueness if provided if data.serial_number: existing_serial = await self.db.execute( select(AssetItem).where( AssetItem.serial_number == data.serial_number, AssetItem.deleted_at.is_(None), ) ) if existing_serial.scalar_one_or_none(): raise AppException( code="SERIAL_NUMBER_EXISTS", message=f"Serial number {data.serial_number} already exists", status_code=400, ) # Resolve brand/model lookup rows + reconcile the denormalised # ``brand`` / ``model`` text columns. The lookup service auto- # creates rows for free-text-only inputs so the intake admin # never has to pre-curate the catalog. from app.domains.asset.brand_model_service import ( AssetBrandModelService, ) bm = AssetBrandModelService(self.db) brand_row, model_row = await bm.resolve_brand_and_model( brand_id=data.brand_id, model_id=data.model_id, brand_name=data.brand, model_name=data.model, ) payload = data.model_dump(exclude_none=True) if brand_row is not None: payload["brand"] = brand_row.name payload["brand_id"] = brand_row.id if model_row is not None: payload["model"] = model_row.name payload["model_id"] = model_row.id # Managed-location gating at create time: drop the field # entirely when the supplied value can't be resolved so the # asset lands without a dirty location string. The dropdown # then asks the admin to set a real one. if "location" in payload: resolver = await self._get_location_resolver() resolved = resolver.resolve(payload["location"]) if resolved is not None: payload["location"] = resolved else: payload.pop("location", None) # Office-held invariant (migration 183). Creating an asset with # ``holder_kind='office'`` short-circuits the personal lifecycle: # the row lands directly at ``status='held_by_office'`` with no # assignee. ``holder_office_id`` is required and must reference # an active office row. holder_kind = payload.get("holder_kind") or ASSET_HOLDER_KIND_PERSONAL if holder_kind == ASSET_HOLDER_KIND_OFFICE: office_id = payload.get("holder_office_id") if office_id is None: raise AppException( code="ASSET_OFFICE_REQUIRED", message=( "holder_kind='office' requires holder_office_id " "pointing at an active asset_office_addresses row." ), status_code=400, ) await self._verify_office_active(office_id) payload["status"] = ASSET_STATUS_HELD_BY_OFFICE payload.pop("assigned_to", None) payload.pop("assigned_at", None) else: payload["holder_kind"] = ASSET_HOLDER_KIND_PERSONAL payload.pop("holder_office_id", None) asset = AssetItem(**payload) self.db.add(asset) # Round 4 audit fix-up 2026-05-04: same IntegrityError shaping # as ``intake_asset`` so create_asset's collision with a # soft-deleted ghost is a friendly 409, not a 500. try: await self.db.flush() # Flush to get the asset ID except IntegrityError as exc: col = _asset_duplicate_unique_column(exc) if col == "asset_tag": raise AppException( code="ASSET_TAG_EXISTS_SOFT_DELETED", message=( f"Asset tag {data.asset_tag} is held by a " "soft-deleted asset. Use a different tag, or " "restore it from the asset recycle bin." ), status_code=409, ) from exc if col == "serial_number": raise AppException( code="SERIAL_NUMBER_EXISTS_SOFT_DELETED", message=( f"Serial number {data.serial_number} is held by " "a soft-deleted asset. Use a different serial, " "or restore it from the asset recycle bin." ), status_code=409, ) from exc raise # Create history entry history = AssetHistory( asset_id=asset.id, action="created", operator_id=operator_id, notes=f"Asset created with tag {asset.asset_tag}", ) self.db.add(history) await self.db.flush() return asset async def get_asset(self, asset_id: UUID) -> AssetItem: """Get asset by ID. Soft-deleted rows are invisible.""" result = await self.db.execute( select(AssetItem) .options( selectinload(AssetItem.history), selectinload(AssetItem.operations), ) .where( AssetItem.id == asset_id, AssetItem.deleted_at.is_(None), ) ) asset = result.scalar_one_or_none() if not asset: raise NotFoundError("Asset", str(asset_id)) return asset async def _get_asset_for_update(self, asset_id: UUID) -> AssetItem: """Load an asset row under ``FOR UPDATE`` for acquire/release mutations (assign / unassign). Without the lock these do a check-then-act on ``asset.status`` with no row lock and no UNIQUE backstop, so two concurrent admin ``POST /assets/{id}/assign`` calls for the same available asset both pass the status check — the second write wins ``assigned_to`` while the first leaves an orphan ``AssetHistory(action='assigned')`` attributing the asset to the loser. Mirrors the locked-load the sibling acquire paths already use (claim/borrow/transfer/ change_assignee). No eager loads — these callers only touch base columns and append a history row. """ result = await self.db.execute( select(AssetItem) .where(AssetItem.id == asset_id, AssetItem.deleted_at.is_(None)) .with_for_update() ) asset = result.scalar_one_or_none() if not asset: raise NotFoundError("Asset", str(asset_id)) return asset async def update_asset( self, asset_id: UUID, data: AssetUpdateRequest, operator_id: UUID, ) -> AssetItem: """Update asset details.""" asset = await self.get_asset(asset_id) if "purchase_entity" in data.model_fields_set and data.purchase_entity is not None: data.purchase_entity = await self._resolve_canonical_purchase_entity_name( data.purchase_entity ) # Check serial number uniqueness if being updated. Mirrors the # `intake_asset` / `create_asset` pattern: probe excludes # soft-deleted ghosts so a friendly ASCII error fires for an # active duplicate, and an IntegrityError catch on flush below # surfaces a 409 if the new serial collides with a soft-deleted # ghost (DB UNIQUE constraint still applies regardless of # deleted_at). if data.serial_number and data.serial_number != asset.serial_number: existing_serial = await self.db.execute( select(AssetItem).where( and_( AssetItem.serial_number == data.serial_number, AssetItem.id != asset_id, AssetItem.deleted_at.is_(None), ) ) ) if existing_serial.scalar_one_or_none(): raise AppException( code="SERIAL_NUMBER_EXISTS", message=f"Serial number {data.serial_number} already exists", status_code=400, ) # Resolve brand/model lookup if either was supplied. Same # auto-create-on-write rules as the intake / create paths. # When only the model is being patched, we MUST keep the # existing brand context — otherwise the lookup helper # would orphan the model under "Unknown" and silently rewrite # the asset's brand text to "Unknown" too. The patch dict # passed in is "fields the user explicitly sent"; everything # else stays as-is on the asset row. patched_kwargs = data.model_dump(exclude_unset=True) sent_brand = "brand" in patched_kwargs or "brand_id" in patched_kwargs sent_model = "model" in patched_kwargs or "model_id" in patched_kwargs if sent_brand or sent_model: from app.domains.asset.brand_model_service import ( AssetBrandModelService, ) bm = AssetBrandModelService(self.db) # When the patch explicitly touches brand/model, drive # resolution from the patched values only. Inheriting # ``asset.brand_id`` here would let resolve_brand_and_model's # "id wins over name" rule silently revert a new ``brand`` # text typed in the edit form (the curated brand row's name # would overwrite "MacBook" back to the existing brand and # the field-level diff below would then see no change). # Only inherit when the field wasn't touched at all. if sent_brand: effective_brand_id = patched_kwargs.get("brand_id") effective_brand_name = patched_kwargs.get("brand") else: effective_brand_id = asset.brand_id effective_brand_name = asset.brand # Same for model. Extra wrinkle: when ONLY the brand is # changing, the inherited ``model_id`` still pins to the # old brand and would re-route ``brand`` back to the old # value inside the resolver (model.brand_id wins over a # contradicting brand_name). Drop the model_id pin in that # case but keep the text so the new brand inherits the # model name via auto-create. if sent_model: effective_model_id = patched_kwargs.get("model_id") effective_model_name = patched_kwargs.get("model") elif sent_brand: effective_model_id = None effective_model_name = asset.model else: effective_model_id = asset.model_id effective_model_name = asset.model brand_row, model_row = await bm.resolve_brand_and_model( brand_id=effective_brand_id, model_id=effective_model_id, brand_name=effective_brand_name, model_name=effective_model_name, ) if brand_row is not None: patched_kwargs["brand"] = brand_row.name patched_kwargs["brand_id"] = brand_row.id if model_row is not None: patched_kwargs["model"] = model_row.name patched_kwargs["model_id"] = model_row.id # Replace data with the resolved fields so the loop # below uses them. data = AssetUpdateRequest(**patched_kwargs) # Managed-location gating on PATCH: when the operator # explicitly set ``location`` (model_fields_set test, NOT a # truthy check \u2014 the empty string is a real "clear it" intent), # resolve the raw value. A resolved hit replaces the patched # value; an unresolvable non-empty value is dropped from the # patch so the existing column doesn't get clobbered with # dirt. An explicit empty / None passes through to clear the # column. # ``dropped_location_raw`` carries the operator's rejected # string into the history entry below so the audit log shows # WHAT they tried to write even when the column itself didn't # change \u2014 Round-1 review medium: prior version was a silent # drop, no user feedback / no audit trace. dropped_location_raw: str | None = None if "location" in data.model_fields_set: raw_location = data.location if raw_location is None or ( isinstance(raw_location, str) and not raw_location.strip() ): # Explicit clear \u2014 leave the assignment loop below # to set ``None``. pass else: resolver = await self._get_location_resolver() resolved = resolver.resolve(raw_location) if resolved is None: # Unresolvable: protect the existing value rather # than blanking it on a typo. The PATCH succeeds # for other fields; the location stays as-is. The # rejected raw value is preserved in # ``dropped_location_raw`` so the AssetHistory # entry below records the operator's intent + a # WARNING goes to the structured logs for ops # visibility. logger.warning( "AssetService.update_asset: dropping unresolvable " "location %r on asset_id=%s; existing value preserved", raw_location, asset_id, ) dropped_location_raw = ( raw_location if isinstance(raw_location, str) else str(raw_location) ) patched_kwargs = data.model_dump(exclude_unset=True) patched_kwargs.pop("location", None) data = AssetUpdateRequest(**patched_kwargs) elif resolved != raw_location: patched_kwargs = data.model_dump(exclude_unset=True) patched_kwargs["location"] = resolved data = AssetUpdateRequest(**patched_kwargs) # Track what changed for history. Split "actual field # changes" from "audit-only entries" (rejected PATCH inputs) # so a typo-only PATCH doesn't bump ``asset.updated_at`` \u2014 # otherwise the asset list page surfaces the row as # "recently updated" even though no column changed # (Round-3 review medium). actual_changes: list[str] = [] for field, value in data.model_dump(exclude_unset=True).items(): old_value = getattr(asset, field) if old_value != value: actual_changes.append(f"{field}: {old_value} \u2192 {value}") setattr(asset, field, value) # Audit-only entries: the column didn't change, but the # operator's rejected attempt is worth recording so the # operations log shows that the PATCH carried a location they # tried and why it wasn't applied. We deliberately do NOT # write the raw value into the history note (Round-4 review): # ``AssetHistory.notes`` is plain text exposed to non-admin # readers via ``GET /assets/{id}`` (in scope) and the # admin-tier-but-not-admin-only ``GET /assets/history`` # endpoint; the raw value is an HRIS-derived # ``work_location_name`` (``Santa Clara`` / ``\u5317\u4eac\u5e02`` / etc.) # and surfacing it to a manager / finance / auditor caller # leaks the same data the operation.detail.location_raw # admin_view gate exists to protect. The raw value still # reaches the ``logger.warning`` output above for admin # correlation via the structured logs \u2014 that's enough for # ops to investigate, without painting the PII onto a # broadly-readable surface. audit_only_notes: list[str] = [] if dropped_location_raw is not None: audit_only_notes.append( "location (rejected, kept prior value) \u2014 submitted value " "did not match the managed asset_locations master; pick " "from the dropdown. Raw value redacted from this " "surface; admins can correlate via the WARNING log." ) # ``updated_at`` is bumped ONLY when a real column changed \u2014 # otherwise the asset list "recently updated" signal lies. if actual_changes: asset.updated_at = datetime.now(UTC) # History row fires for either kind \u2014 admins see the audit # trail even on rejected PATCHes, but the asset row's # freshness signal stays honest. if actual_changes or audit_only_notes: history = AssetHistory( asset_id=asset.id, action="updated", operator_id=operator_id, notes=f"Updated: {'; '.join(actual_changes + audit_only_notes)}", ) self.db.add(history) try: await self.db.flush() except IntegrityError as exc: if _asset_duplicate_unique_column(exc) == "serial_number": raise AppException( code="SERIAL_NUMBER_EXISTS_SOFT_DELETED", message=( f"Serial number {data.serial_number} is held by " "another (possibly soft-deleted) asset. Use a " "different serial, or restore it from the asset " "recycle bin." ), status_code=409, ) from exc raise return asset # Promoted to ``app.domains.asset.models.ASSET_IN_USE_STATUSES`` # (Round 4 audit fix-up 2026-05-04) so the offboarding domain's # ``_rescan`` blocker scan uses the same canonical set. Class-level # alias kept for back-compat with internal callers. _IN_USE_STATUSES = ASSET_IN_USE_STATUSES async def delete_asset( self, asset_id: UUID, operator_id: UUID, ) -> AssetItem: """Soft-delete an asset. Refuses if the asset is in use by anyone (assigned, claimed, borrowed, or any pending state). Soft-delete preserves the ``asset_history`` and ``asset_operations`` rows so the audit chain remains intact (CLAUDE.md rule 6 — audit by design). Pre-launch hardening 2026-05-04 replaced the prior hard-DELETE path which cascade-deleted history/operations and made forensic queries against the audit log return dangling resource IDs. """ asset = await self.get_asset(asset_id) # Explicit held_by_leaver guard — relying on the # ``assigned_to is not None`` short-circuit was fragile: if a # future fix accidentally clears ``assigned_to`` on a # held_by_leaver row (the documented invariant says we MUST # preserve it), this fall-through would let an admin DELETE the # asset and wipe the leaver-holding history with it. if asset.status == ASSET_STATUS_HELD_BY_LEAVER: raise AppException( code="ASSET_HELD_BY_LEAVER", message=( "Cannot delete an asset still held by a departed " "employee. Run /assets/{id}/recover-from-leaver first." ), status_code=409, ) if ( asset.assigned_to is not None or asset.borrowed_by is not None or asset.status in self._IN_USE_STATUSES ): raise AppException( code="ASSET_IN_USE", message="Cannot delete an asset that is in use. Return or reclaim it first.", status_code=400, ) # Reserved-for-a-hire guard (2026-09-05). A reserved device sits # in a STOCK status, so every check above waves it through — and # the soft delete then made the earmark vanish silently: the # hardware-prep worklist stopped listing it (that query filters # ``deleted_at IS NULL``), the hire showed as un-prepped again # with no trace of why, and since 2026-09-05 an earmark also # satisfies the Day-1 hardware dimension, so the dimension # flipped back to pending with no event explaining it. Refuse and # name the release action instead. if asset.reserved_for_person_id is not None: raise AppException( code="ASSET_RESERVED_FOR_HIRE", message=( "Cannot delete an asset earmarked for an incoming " "hire. Release the reservation first " "(/assets/hardware-prep/reservations/{asset_id})." ), status_code=409, ) now = datetime.now(UTC) asset.deleted_at = now asset.updated_at = now history = AssetHistory( asset_id=asset.id, action="deleted", operator_id=operator_id, notes="Asset soft-deleted", ) self.db.add(history) await self.db.flush() return asset async def restore_asset( self, asset_id: UUID, operator_id: UUID, ) -> AssetItem: """Restore a soft-deleted asset from the recycle bin. Clears ``deleted_at`` and writes a ``restored`` history row. Admin-only at the router. Refuses (409) if the asset isn't actually deleted so a double-click / stale list doesn't silently no-op. No unique-collision handling is needed: the ``asset_tag`` / ``serial_number`` UNIQUE indexes cover ALL rows (incl. soft-deleted), so while this ghost held the tag/serial no live row could have taken it — restoring it back can't collide. """ asset = await self.db.get(AssetItem, asset_id) if asset is None: raise NotFoundError("Asset", str(asset_id)) if asset.deleted_at is None: raise AppException( code="ASSET_NOT_DELETED", message="Asset is not deleted; nothing to restore.", status_code=409, ) now = datetime.now(UTC) asset.deleted_at = None asset.updated_at = now history = AssetHistory( asset_id=asset.id, action="restored", operator_id=operator_id, notes="Asset restored from recycle bin", ) self.db.add(history) await self.db.flush() return asset async def list_deleted_assets( self, page: int = 1, page_size: int = 20, search: str | None = None, ) -> tuple[list[AssetItem], int]: """List soft-deleted assets for the admin recycle bin, newest deletion first. Admin-only at the router. ``search`` matches asset tag / name / serial number (LIKE, wildcard-escaped).""" base = select(AssetItem).where(AssetItem.deleted_at.is_not(None)) count_query = ( select(func.count()) .select_from(AssetItem) .where(AssetItem.deleted_at.is_not(None)) ) if search and search.strip(): pattern = f"%{_esc(search.strip())}%" cond = or_( AssetItem.asset_tag.ilike(pattern), AssetItem.asset_name.ilike(pattern), AssetItem.serial_number.ilike(pattern), ) base = base.where(cond) count_query = count_query.where(cond) total = (await self.db.execute(count_query)).scalar() or 0 offset = (page - 1) * page_size query = ( base.order_by(desc(AssetItem.deleted_at)) .offset(offset) .limit(page_size) ) assets = list((await self.db.execute(query)).scalars().all()) return assets, total def _ensure_not_reserved_for_other( self, asset: AssetItem, target_person_id: UUID, *, clear_on_match: bool = True, ) -> None: """Reservation guard (migration 279, 备机闭环). An asset earmarked for an incoming hire must not be handed to anyone else through any grant path (assign / claim / borrow / transfer / change_assignee). Granting it to the reserved person fulfils the reservation — cleared in the same write unless the operation doesn't change ownership (borrow keeps it). """ if asset.reserved_for_person_id is None: return if asset.reserved_for_person_id != target_person_id: raise AppException( code="ASSET_RESERVED", message=( "Asset is reserved for an incoming hire; release the " "reservation first (资产工作台 → 待备机)" ), status_code=409, ) if clear_on_match: asset.reserved_for_person_id = None asset.reserved_at = None async def assign_asset( self, asset_id: UUID, request: AssetAssignRequest, operator_id: UUID, ) -> AssetItem: """Assign asset to a user.""" asset = await self._get_asset_for_update(asset_id) if asset.status not in ["idle", "available"]: raise AppException( code="ASSET_NOT_AVAILABLE", message=f"Asset is {asset.status}, cannot assign", status_code=400, ) self._ensure_not_reserved_for_other(asset, request.user_id) # Round-49: route through ``_verify_user`` so soft-deleted and # leaver_pending/left users are refused. The previous direct # query was the surviving member of the leaver-gap family — # Round 44 closed claim/borrow/change_assignee, Round 45 closed # transfer; this branch let asset admins dispatch hardware to # leavers via ``POST /assets/{id}/assign``. user = await self._verify_user(request.user_id) # Update asset asset.assigned_to = request.user_id asset.assigned_at = datetime.now(UTC) asset.status = "assigned" asset.updated_at = datetime.now(UTC) # Create history entry history = AssetHistory( asset_id=asset.id, action="assigned", to_user=request.user_id, operator_id=operator_id, notes=request.notes or f"Assigned to {user.display_name or user.email}", ) self.db.add(history) await self.db.flush() return asset async def unassign_asset( self, asset_id: UUID, operator_id: UUID, notes: str | None = None, ) -> AssetItem: """Unassign (reclaim) an asset.""" asset = await self._get_asset_for_update(asset_id) if asset.status != "assigned": raise AppException( code="ASSET_NOT_ASSIGNED", message="Asset is not currently assigned", status_code=400, ) # Create history entry history = AssetHistory( asset_id=asset.id, action="unassigned", from_user=asset.assigned_to, operator_id=operator_id, notes=notes or "Asset reclaimed", ) self.db.add(history) # Update asset. Lands on ``idle``, not the legacy ``available`` # synonym: this was the last live writer of a second token meaning # "in stock with no holder", and one meaning with two tokens is # how the fleet ends up showing 「可用」 next to 「闲置」 for # identical rows (and how ``in_stock`` hid for months — see # migration 454). ``available`` stays *accepted* by every gate for # pre-existing rows; nothing writes it any more. asset.assigned_to = None asset.assigned_at = None asset.status = "idle" asset.updated_at = datetime.now(UTC) await self.db.flush() return asset async def transfer_asset( self, asset_id: UUID, request: AssetTransferRequest, operator_id: UUID, ) -> AssetItem: """Transfer asset from one user to another.""" result = await self.db.execute( select(AssetItem).where(AssetItem.id == asset_id, AssetItem.deleted_at.is_(None)).with_for_update() ) asset = result.scalar_one_or_none() if not asset: raise NotFoundError("Asset", str(asset_id)) if asset.status not in ("assigned", "claimed"): raise AppException( code="ASSET_NOT_ASSIGNED", message="Asset is not currently assigned", status_code=400, ) if asset.assigned_to == request.to_user_id: raise AppException( code="SAME_USER", message="Asset is already assigned to this user", status_code=400, ) # Pre-launch P0 fix (Round 45, CR-404): verify recipient is an # active employee (not soft-deleted, not leaver_pending, not # left). Without this, an asset admin could transfer hardware # to a leaver via /assets/{id}/transfer — the only ownership # transition path that bypassed _verify_user (Round-44 closed # claim / borrow / change_assignee but missed transfer). user = await self._verify_user(request.to_user_id) self._ensure_not_reserved_for_other(asset, request.to_user_id) from_user_id = asset.assigned_to notes_text = request.notes or ( f"Transferred to {user.display_name or user.email}" ) # Create history entry history = AssetHistory( asset_id=asset.id, action="transferred", from_user=from_user_id, to_user=request.to_user_id, operator_id=operator_id, notes=notes_text, ) self.db.add(history) # Mirror ``change_assignee`` — the AssetOperation row is what # the admin operations page renders. Without it, transfers via # this endpoint were invisible there while ``change_assignee`` # transfers were visible, producing inconsistent admin UX. operation = AssetOperation( operation_number=_generate_operation_number(), asset_id=asset.id, operation_type="transfer", status="confirmed", from_user_id=from_user_id, to_user_id=request.to_user_id, operator_id=operator_id, notes=notes_text, confirmed_at=datetime.now(UTC), confirmed_by=operator_id, ) self.db.add(operation) # Update asset asset.assigned_to = request.to_user_id asset.assigned_at = datetime.now(UTC) asset.updated_at = datetime.now(UTC) await self.db.flush() return asset # Whitelist of allowed sort fields → SQLAlchemy column mapping ALLOWED_SORT_FIELDS: dict[str, Any] = { "created_at": AssetItem.created_at, "updated_at": AssetItem.updated_at, "asset_tag": AssetItem.asset_tag, "asset_name": AssetItem.asset_name, "category": AssetItem.category, "status": AssetItem.status, "device_status": AssetItem.device_status, "purchase_date": AssetItem.purchase_date, "purchase_cost": AssetItem.purchase_cost, "purchase_entity": AssetItem.purchase_entity, "usage_company": AssetItem.usage_company, "currency": AssetItem.currency, "assigned_at": AssetItem.assigned_at, "expected_return_date": AssetItem.expected_return_date, "warranty_until": AssetItem.warranty_until, "location": AssetItem.location, } # Nullable sort fields need NULLS LAST handling NULLABLE_SORT_FIELDS = { "asset_name", "purchase_date", "purchase_cost", "purchase_entity", "usage_company", "currency", "assigned_at", "expected_return_date", "warranty_until", "location", } async def list_assets( self, page: int = 1, page_size: int = 20, status: list[str] | None = None, category: list[str] | None = None, location: list[str] | None = None, usage_company: list[str] | None = None, purchase_entity: list[str] | None = None, assigned_to: UUID | None = None, assigned_to_ids: list[UUID] | None = None, holder_kind: list[str] | None = None, holder_office_id: list[UUID] | None = None, search: str | None = None, sort_by: str = "created_at", sort_order: str = "desc", ) -> tuple[list[AssetItem], int]: """List assets with filters, sorting, and pagination. Soft-deleted rows (``deleted_at IS NOT NULL``) are excluded from every list query — admins do not see them and they cannot be re-edited. """ # Two LEFT JOINs into ``identity_persons`` so a free-text # ``search`` can match the current assignee OR the current # borrower (display_name / work_email / employee_no). Without # these joins admins searching by holder name can't find # borrowed laptops — bug reported 2026-05-09. assignee_person = aliased(AuthUser) borrower_person = aliased(AuthUser) query = ( select(AssetItem) .outerjoin(assignee_person, AssetItem.assigned_to == assignee_person.id) .outerjoin(borrower_person, AssetItem.borrowed_by == borrower_person.id) .where(AssetItem.deleted_at.is_(None)) ) count_query = ( select(func.count(func.distinct(AssetItem.id))) .select_from(AssetItem) .outerjoin(assignee_person, AssetItem.assigned_to == assignee_person.id) .outerjoin(borrower_person, AssetItem.borrowed_by == borrower_person.id) .where(AssetItem.deleted_at.is_(None)) ) # Apply filters (support multi-value via IN) conditions = [] if status: conditions.append(AssetItem.status.in_(status)) if category: conditions.append(AssetItem.category.in_(category)) if location: conditions.append(AssetItem.location.in_(location)) if usage_company: conditions.append(AssetItem.usage_company.in_(usage_company)) if purchase_entity: conditions.append(AssetItem.purchase_entity.in_(purchase_entity)) # Match physical-custody semantics: a person "has" an asset if # they own it (assigned_to) OR currently hold it on loan # (borrowed_by). This mirrors the free-text search predicate # below + ``get_user_assets`` (used by /my/inventory). Without # the OR a "filter by 使用人 = X" call returns fewer rows than # "search 'X'" against the same dataset — confusingly different # answers from the same page (reported 2026-05-09: filter # showed 3, search showed 4 for the same person who had 3 owned # + 1 borrowed laptop). if assigned_to: conditions.append( or_( AssetItem.assigned_to == assigned_to, AssetItem.borrowed_by == assigned_to, ) ) if assigned_to_ids is not None: conditions.append( or_( AssetItem.assigned_to.in_(assigned_to_ids), AssetItem.borrowed_by.in_(assigned_to_ids), ) ) # Holder-kind / office filters (PR 2026-05-26). Office-held rows # (migration 183) live in the same ``asset_items`` table as # personal-held rows — the only differentiator is # ``holder_kind`` + ``holder_office_id``. Without these filters # admins have no way to scope the list to "all office-held" or # "Shanghai office only" — the previous workaround (search by # location string) misses because office name and location text # are not the same column. Validation: reject unknown # ``holder_kind`` values at the service boundary so MCP / API / # internal callers all share the same contract. if holder_kind: invalid = [k for k in holder_kind if k not in ( ASSET_HOLDER_KIND_PERSONAL, ASSET_HOLDER_KIND_OFFICE, )] if invalid: raise ValidationError( f"unknown holder_kind value(s): {invalid!r}; " f"allowed: {[ASSET_HOLDER_KIND_PERSONAL, ASSET_HOLDER_KIND_OFFICE]}", ) conditions.append(AssetItem.holder_kind.in_(holder_kind)) if holder_office_id: conditions.append(AssetItem.holder_office_id.in_(holder_office_id)) if search: pattern = f"%{_esc(search)}%" conditions.append( or_( AssetItem.asset_tag.ilike(pattern), AssetItem.brand.ilike(pattern), AssetItem.model.ilike(pattern), AssetItem.serial_number.ilike(pattern), AssetItem.asset_name.ilike(pattern), assignee_person.display_name.ilike(pattern), assignee_person.work_email.ilike(pattern), assignee_person.employee_no.ilike(pattern), borrower_person.display_name.ilike(pattern), borrower_person.work_email.ilike(pattern), borrower_person.employee_no.ilike(pattern), ) ) if conditions: query = query.where(and_(*conditions)) count_query = count_query.where(and_(*conditions)) # Get total count total_result = await self.db.execute(count_query) total = total_result.scalar() or 0 # Apply sorting sort_column = self.ALLOWED_SORT_FIELDS[sort_by] order_fn = asc if sort_order == "asc" else desc order_clause = order_fn(sort_column) order_by_clauses = [order_clause] if sort_by in self.NULLABLE_SORT_FIELDS: # MySQL/TiDB do not support "NULLS LAST", so emulate it explicitly. order_by_clauses = [sort_column.is_(None).asc(), order_clause] # Apply pagination offset = (page - 1) * page_size query = query.offset(offset).limit(page_size).order_by(*order_by_clauses) # Execute query result = await self.db.execute(query) assets = list(result.scalars().all()) return assets, total async def get_user_assets(self, user_id: UUID) -> list[AssetItem]: """Get all assets a user holds — owned OR currently borrowed. Includes: * assets assigned to the user (``assigned_to == user_id``) * assets currently on loan to the user (``borrowed_by == user_id`` AND status='borrowed') The borrow-side OR catches "physical custody" so /my/inventory accurately reflects what the user has in hand, not just what they own. """ result = await self.db.execute( select(AssetItem) .where( AssetItem.deleted_at.is_(None), or_( AssetItem.assigned_to == user_id, and_( AssetItem.borrowed_by == user_id, AssetItem.status == "borrowed", ), ), ) .order_by(AssetItem.assigned_at.desc()) ) return list(result.scalars().all()) async def get_subordinate_assets( self, manager_id: UUID, ) -> list[AssetItem]: """Assets held by manager's direct + transitive reports. BFS-walks ``IdentityPerson.manager_person_id`` so skip-level reports (A → B → C) are included when A asks for "team assets". The previous direct-only ``WHERE manager_person_id == manager_id`` query missed every report below the first level. Excludes the manager themselves and anyone with ``employment_status IN ('left', 'leaver_pending')`` (the ``ACTIVE_ONLY`` scope of the canonical resolver). """ visited: set[UUID] = set( await subordinate_person_ids( self.db, manager_id, scope=SubordinateScope.ACTIVE_ONLY, ) ) if not visited: return [] result = await self.db.execute( select(AssetItem) .where( AssetItem.deleted_at.is_(None), or_( AssetItem.assigned_to.in_(visited), and_( AssetItem.borrowed_by.in_(visited), AssetItem.status == "borrowed", ), ), ) .order_by(AssetItem.assigned_at.desc()) ) return list(result.scalars().all()) async def get_assigned_assets(self) -> list[AssetItem]: """Get all assets currently assigned to a person.""" result = await self.db.execute( select(AssetItem) .where( AssetItem.assigned_to.is_not(None), AssetItem.deleted_at.is_(None), ) .order_by(AssetItem.assigned_at.desc(), AssetItem.created_at.desc()) ) return list(result.scalars().all()) async def get_asset_history( self, asset_id: UUID | None = None, page: int = 1, page_size: int = 20, ) -> tuple[list[AssetHistory], int]: """Get asset history, optionally filtered by asset.""" query = select(AssetHistory) count_query = select(func.count()).select_from(AssetHistory) if asset_id: query = query.where(AssetHistory.asset_id == asset_id) count_query = count_query.where(AssetHistory.asset_id == asset_id) # Get total count total_result = await self.db.execute(count_query) total = total_result.scalar() or 0 # Apply pagination offset = (page - 1) * page_size query = query.offset(offset).limit(page_size).order_by(AssetHistory.created_at.desc()) # Execute query result = await self.db.execute(query) history = list(result.scalars().all()) return history, total async def get_stats(self) -> AssetStatsResponse: """Get asset statistics.""" # Total assets (exclude soft-deleted / disposed) total_result = await self.db.execute( select(func.count()).select_from(AssetItem) .where(AssetItem.deleted_at.is_(None)) ) total_assets = total_result.scalar() or 0 # By status (exclude soft-deleted so headline total == sum of statuses) status_result = await self.db.execute( select(AssetItem.status, func.count()) .where(AssetItem.deleted_at.is_(None)) .group_by(AssetItem.status) ) by_status = dict(status_result.fetchall()) # By category category_result = await self.db.execute( select(AssetItem.category, func.count()) .where(AssetItem.deleted_at.is_(None)) .group_by(AssetItem.category) ) by_category = dict(category_result.fetchall()) # By brand (excluding NULLs / blanks) brand_result = await self.db.execute( select(AssetItem.brand, func.count()) .where( AssetItem.deleted_at.is_(None), AssetItem.brand.isnot(None), AssetItem.brand != "", ) .group_by(AssetItem.brand) .order_by(func.count().desc()) ) by_brand = dict(brand_result.fetchall()) # By location (excluding NULLs / blanks) location_result = await self.db.execute( select(AssetItem.location, func.count()) .where( AssetItem.deleted_at.is_(None), AssetItem.location.isnot(None), AssetItem.location != "", ) .group_by(AssetItem.location) .order_by(func.count().desc()) ) by_location = dict(location_result.fetchall()) # By usage company / 使用主体 (excluding NULLs / blanks) company_result = await self.db.execute( select(AssetItem.usage_company, func.count()) .where( AssetItem.deleted_at.is_(None), AssetItem.usage_company.isnot(None), AssetItem.usage_company != "", ) .group_by(AssetItem.usage_company) .order_by(func.count().desc()) ) by_usage_company = dict(company_result.fetchall()) # By model (top 10 — same exclusions as brand) model_result = await self.db.execute( select(AssetItem.model, func.count()) .where( AssetItem.deleted_at.is_(None), AssetItem.model.isnot(None), AssetItem.model != "", ) .group_by(AssetItem.model) .order_by(func.count().desc()) .limit(10) ) by_model = dict(model_result.fetchall()) # Geographic distribution — resolve each asset to a country/city via # its holding office (office-held mode) first, else the assigned # person's location. Normalized to an ISO alpha-2 country code so the # frontend world map keys consistently with the HR people-geo endpoint. from app.core.geo import normalize_country_code # noqa: PLC0415 from app.identity.person_models import IdentityPerson # noqa: PLC0415 office = aliased(AssetOfficeAddress) holder = aliased(IdentityPerson) country_expr = func.coalesce(office.country_or_region, holder.country) city_expr = func.coalesce(office.city, holder.city) geo_result = await self.db.execute( select(country_expr, city_expr, func.count()) .select_from(AssetItem) .outerjoin(office, AssetItem.holder_office_id == office.id) .outerjoin(holder, AssetItem.assigned_to == holder.id) .where(AssetItem.deleted_at.is_(None)) .group_by(country_expr, city_expr) ) by_country: dict[str, int] = {} by_china_city: dict[str, int] = {} for country_raw, city_raw, cnt in geo_result.all(): n = int(cnt) code = normalize_country_code(country_raw) if code: by_country[code] = by_country.get(code, 0) + n if code == "CN": city = (city_raw or "").strip() or "unknown" by_china_city[city] = by_china_city.get(city, 0) + n # Total value (exclude soft-deleted/disposed assets, matching the other # stats above — otherwise total_value can exceed the on-register total). value_result = await self.db.execute( select(func.sum(AssetItem.purchase_cost)) .where(AssetItem.deleted_at.is_(None)) ) total_value = value_result.scalar() or Decimal("0") # Assigned count assigned_result = await self.db.execute( select(func.count()).select_from(AssetItem) .where(AssetItem.status == "assigned", AssetItem.deleted_at.is_(None)) ) assigned_count = assigned_result.scalar() or 0 # Available count available_result = await self.db.execute( select(func.count()).select_from(AssetItem) .where(AssetItem.status.in_(["available", "idle"]), AssetItem.deleted_at.is_(None)) ) available_count = available_result.scalar() or 0 # Warranty expiring soon (next 30 days) today = date.today() thirty_days = today + timedelta(days=30) warranty_result = await self.db.execute( select(func.count()).select_from(AssetItem) .where( and_( AssetItem.warranty_until.isnot(None), AssetItem.warranty_until >= today, AssetItem.warranty_until <= thirty_days, AssetItem.deleted_at.is_(None), ) ) ) warranty_expiring_soon = warranty_result.scalar() or 0 # Recently added (last 30 days) thirty_days_ago = datetime.now(UTC) - timedelta(days=30) recent_result = await self.db.execute( select(func.count()).select_from(AssetItem) .where(AssetItem.created_at >= thirty_days_ago, AssetItem.deleted_at.is_(None)) ) recently_added = recent_result.scalar() or 0 return AssetStatsResponse( total_assets=total_assets, by_status=by_status, by_category=by_category, by_brand=by_brand, by_location=by_location, by_usage_company=by_usage_company, by_model=by_model, by_country=by_country, by_china_city=by_china_city, total_value=total_value, assigned_count=assigned_count, available_count=available_count, warranty_expiring_soon=warranty_expiring_soon, recently_added=recently_added, ) # ------------------------------------------------------------------ # New lifecycle operations # ------------------------------------------------------------------ async def _verify_user(self, user_id: UUID) -> AuthUser: """Verify a user exists, is not soft-deleted, and is an active employee. Pre-launch hardening 2026-05-04: rejects ``leaver_pending`` and ``left`` employment states. Previously asset claim / borrow / change-assignee silently dispatched hardware to leavers because only the SSO suspension path checked employment_status — leaving a window where IT could hand a laptop to someone whose offboarding had already started. """ result = await self.db.execute( select(AuthUser).where( AuthUser.id == user_id, AuthUser.deleted_at.is_(None), ) ) user = result.scalar_one_or_none() if not user: raise NotFoundError("User", str(user_id)) # Block leavers from being grant targets. The check is permissive # of NULL (legacy rows / sync gap) — only an explicit non-active # state rejects. Direct attribute access — IdentityPerson always # has employment_status (memory: feedback_no_getattr_default). status = (user.employment_status or "").lower() if status in {"leaver_pending", "left"}: raise AppException( code="USER_NOT_ACTIVE", message=( f"User {user.display_name or user.email} is " f"{status}; cannot receive resources." ), status_code=409, ) return user async def intake_asset( self, data: AssetIntakeRequest, operator_id: UUID, ) -> tuple[AssetItem, AssetOperation]: """Register/intake a new asset and create an intake operation.""" # Resolve ``purchase_entity_id`` (preferred) → ``purchase_entity`` # name. Single source of truth: the ``asset_legal_entities`` row # the admin selected, not whatever they typed. Free-text writes # to ``purchase_entity`` are still honoured for backward compat # but the structured FK-shaped path always wins. resolved_entity: AssetLegalEntity | None = None if data.purchase_entity_id is not None: resolved_entity = await self.db.get( AssetLegalEntity, data.purchase_entity_id, ) if resolved_entity is None: raise NotFoundError( "Purchase entity", str(data.purchase_entity_id), ) if not resolved_entity.is_active: raise AppException( code="LEGAL_ENTITY_INACTIVE", message="Purchase entity is inactive", status_code=400, ) data.purchase_entity = resolved_entity.name elif data.purchase_entity is not None: data.purchase_entity = await self._resolve_canonical_purchase_entity_name( data.purchase_entity ) # Auto-generate the asset tag when the admin didn't provide one # (the new intake form shipped 2026-05-08 hides the field). Tag # format: ``IT-{entity.code}-{code}-{YYYYMMDDHHMMSS}``. The # ``{code}`` segment comes from the ``asset_category_configs`` row # whose ``asset_category`` matches the asset category — admins manage # it via the 分类管理 tab (no longer a hardcoded mapping). if not data.asset_tag: if resolved_entity is None: raise AppException( code="PURCHASE_ENTITY_REQUIRED", message="purchase_entity_id is required to auto-generate " "an asset tag", status_code=400, ) data.asset_tag = await self.generate_asset_tag_for_category( purchase_entity_id=resolved_entity.id, category=data.category, ) # Check if asset_tag already exists among ACTIVE rows. Soft-deleted # collisions are handled separately below: the flush hits the # soft-delete-ignoring DB UNIQUE, the IntegrityError is caught, and # a friendly 409 ``ASSET_TAG_EXISTS_SOFT_DELETED`` is surfaced. So # this guard intentionally filters ``deleted_at IS NULL`` — an # active collision is a 400, a soft-deleted ghost is a 409. existing = await self.db.execute( select(AssetItem).where( AssetItem.asset_tag == data.asset_tag, AssetItem.deleted_at.is_(None), ) ) if existing.scalar_one_or_none(): raise AppException( code="ASSET_TAG_EXISTS", message=f"Asset tag {data.asset_tag} already exists", status_code=400, ) if data.serial_number: existing_serial = await self.db.execute( select(AssetItem).where( AssetItem.serial_number == data.serial_number, AssetItem.deleted_at.is_(None), ) ) if existing_serial.scalar_one_or_none(): raise AppException( code="SERIAL_NUMBER_EXISTS", message=f"Serial number {data.serial_number} already exists", status_code=400, ) # Resolve brand/model lookup (same pattern as create_asset). from app.domains.asset.brand_model_service import ( AssetBrandModelService, ) bm = AssetBrandModelService(self.db) brand_row, model_row = await bm.resolve_brand_and_model( brand_id=data.brand_id, model_id=data.model_id, brand_name=data.brand, model_name=data.model, ) asset_data = data.model_dump(exclude_none=True) asset_data["status"] = "idle" # ``purchase_entity_id`` is a transient input on the intake # schema (used above to resolve + stamp the entity name) — not # a column on ``asset_items``. Strip it before the model # constructor or SQLAlchemy will raise ``TypeError: invalid # keyword argument``. asset_data.pop("purchase_entity_id", None) # Office-held intake (migration 183). When ``holder_kind=office`` # the asset skips the idle → claim path and lands directly at # ``status='held_by_office'`` under the named office. Same # invariant as ``create_asset``: holder_office_id required, # office must be active, assigned_to is cleared. holder_kind = asset_data.get( "holder_kind", ASSET_HOLDER_KIND_PERSONAL, ) if holder_kind == ASSET_HOLDER_KIND_OFFICE: office_id = asset_data.get("holder_office_id") if office_id is None: raise AppException( code="ASSET_OFFICE_REQUIRED", message=( "holder_kind='office' requires holder_office_id" ), status_code=400, ) await self._verify_office_active(office_id) asset_data["status"] = ASSET_STATUS_HELD_BY_OFFICE asset_data.pop("assigned_to", None) asset_data.pop("assigned_at", None) else: asset_data["holder_kind"] = ASSET_HOLDER_KIND_PERSONAL asset_data.pop("holder_office_id", None) if brand_row is not None: asset_data["brand"] = brand_row.name asset_data["brand_id"] = brand_row.id if model_row is not None: asset_data["model"] = model_row.name asset_data["model_id"] = model_row.id # Managed-location gating at intake. Mirrors ``create_asset``: # drop unresolvable raw values rather than letting them land # as free-text — otherwise the field's "single source of truth # is the master" contract loses meaning at the most common # write path. if "location" in asset_data: resolver = await self._get_location_resolver() resolved = resolver.resolve(asset_data["location"]) if resolved is not None: asset_data["location"] = resolved else: asset_data.pop("location", None) asset = AssetItem(**asset_data) self.db.add(asset) # Round 4 audit fix-up 2026-05-04: catch IntegrityError so a # collision with a soft-deleted asset's tag/serial surfaces as # a friendly 409 instead of a raw 500. The service-layer probe # already filters ``deleted_at IS NULL`` (so the friendly # ASSET_TAG_EXISTS doesn't fire for a soft-deleted ghost), but # the DB UNIQUE constraint still applies and would otherwise # bubble as an unhandled IntegrityError. try: await self.db.flush() except IntegrityError as exc: # ``_asset_duplicate_unique_column`` gates on MySQL/TiDB error # 1062 (DUPLICATE KEY) and extracts the colliding column from # the index name, so an unrelated constraint violation isn't # misclassified and the prod ``uk_asset_items_*`` index name is # matched (not just the test-only column-named index). col = _asset_duplicate_unique_column(exc) if col == "asset_tag": raise AppException( code="ASSET_TAG_EXISTS_SOFT_DELETED", message=( f"Asset tag {data.asset_tag} is held by a " "soft-deleted asset. Use a different tag, or " "restore it from the asset recycle bin." ), status_code=409, ) from exc if col == "serial_number": raise AppException( code="SERIAL_NUMBER_EXISTS_SOFT_DELETED", message=( f"Serial number {data.serial_number} is held by " "a soft-deleted asset. Use a different serial, " "or restore it from the asset recycle bin." ), status_code=409, ) from exc raise # Create operation record operation = AssetOperation( operation_number=_generate_operation_number(), asset_id=asset.id, operation_type="intake", status="confirmed", operator_id=operator_id, notes=data.notes or f"Asset intake: {asset.asset_tag}", confirmed_at=datetime.now(UTC), confirmed_by=operator_id, ) self.db.add(operation) # Also create legacy history history = AssetHistory( asset_id=asset.id, action="created", operator_id=operator_id, notes=f"Asset intake with tag {asset.asset_tag}", ) self.db.add(history) await self.db.flush() return asset, operation async def claim_asset( self, asset_id: UUID, request: AssetClaimRequest, operator_id: UUID, ) -> tuple[AssetItem, AssetOperation]: """Claim/dispatch an asset to a user.""" result = await self.db.execute( select(AssetItem).where(AssetItem.id == asset_id, AssetItem.deleted_at.is_(None)).with_for_update() ) asset = result.scalar_one_or_none() if not asset: raise NotFoundError("Asset", str(asset_id)) # Office-held assets are a legitimate claim source (migration # 183) — an employee can pick up a meeting-room laptop or # shared dock and it transitions out of office_held into # personal claimed. The release of the office_held state # happens here directly, not via a separate # release_from_office call, so the operation log records a # single claim event rather than two (release + claim). if asset.status not in ( "idle", "available", ASSET_STATUS_HELD_BY_OFFICE, ): raise AppException( code="ASSET_NOT_AVAILABLE", message=f"Asset is {asset.status}, cannot claim", status_code=400, ) user = await self._verify_user(request.to_user_id) self._ensure_not_reserved_for_other(asset, request.to_user_id) # Update asset with claim details asset.assigned_to = request.to_user_id asset.assigned_at = datetime.now(UTC) asset.status = "claimed" # Claiming flips a previously office-held device into a # personal-held one — clear the office pointer so inventory # routing now goes to the new assignee, not the office admin. asset.holder_kind = ASSET_HOLDER_KIND_PERSONAL asset.holder_office_id = None asset.updated_at = datetime.now(UTC) # Update optional fields from request. When not provided, sync from assignee contract entity. resolved_usage_company = request.usage_company or user.company_name if resolved_usage_company: asset.usage_company = resolved_usage_company # Route the location through the managed-location resolver so # raw HRIS / form values (``Santa Clara``, ``北京市``, # ``Würzburg``) land as canonical ``asset_locations.name`` # values — or as None when the value can't be resolved. The # detail dict below records BOTH the resolved value and the # original raw input when they differ, so audit shows what # the operator submitted vs. what was stored. location_resolver = await self._get_location_resolver() resolved_location = location_resolver.resolve(request.location) if resolved_location is not None: asset.location = resolved_location # Store extra fields in operation detail detail = {} if request.claim_date: detail["claim_date"] = request.claim_date.isoformat() else: detail["claim_date"] = date.today().isoformat() if resolved_usage_company: detail["usage_company"] = resolved_usage_company if request.usage_department: detail["usage_department"] = request.usage_department if resolved_location is not None: detail["location"] = resolved_location if request.location and resolved_location != request.location: # Captures the resolver in action: a HRIS-derived # ``Santa Clara`` arrived but the asset row now stores # ``美国`` (or None). Persisting the raw value makes the # resolver's behavior auditable from the operation log. detail["location_raw"] = request.location operation = AssetOperation( operation_number=_generate_operation_number(), asset_id=asset.id, operation_type="claim", status="confirmed", to_user_id=request.to_user_id, operator_id=operator_id, notes=request.notes or f"Claimed by {user.display_name or user.email}", detail=detail if detail else None, confirmed_at=datetime.now(UTC), confirmed_by=operator_id, ) self.db.add(operation) history = AssetHistory( asset_id=asset.id, action="assigned", to_user=request.to_user_id, operator_id=operator_id, notes=request.notes or f"Claimed by {user.display_name or user.email}", ) self.db.add(history) await self.db.flush() return asset, operation async def return_asset( self, asset_id: UUID, request: AssetReturnRequest, operator_id: UUID, ) -> tuple[AssetItem, AssetOperation]: """Return an asset to stock.""" result = await self.db.execute( select(AssetItem).where(AssetItem.id == asset_id, AssetItem.deleted_at.is_(None)).with_for_update() ) asset = result.scalar_one_or_none() if not asset: raise NotFoundError("Asset", str(asset_id)) if asset.status not in ("claimed", "assigned"): raise AppException( code="ASSET_NOT_CLAIMED", message=f"Asset is {asset.status}, cannot return", status_code=400, ) from_user_id = asset.assigned_to # Resolve return_target (migration 183, 2026-05-19). # Default behaviour (pre-183): every return lands the asset back # in ``status='idle'``. Office-held mode adds a "return to my # home office" target: the device becomes office-held under the # returner's work-location office, so the next inventory cycle # routes to that office's admin rather than IT fallback. # # Resolution order: # 1. ``request.return_target`` explicit value wins. # 2. If 'office' but no target_office_id, try the returner's # home office via ``resolve_office_for_person``. # 3. If still no office, fall back to idle (safer than # silently dropping the return into the wrong office). return_target = request.return_target resolved_office: AssetOfficeAddress | None = None if return_target == "office": if request.target_office_id is not None: resolved_office = await self._verify_office_active( request.target_office_id, ) elif from_user_id is not None: resolved_office = await self.resolve_office_for_person( from_user_id, ) if resolved_office is None: # No usable office — degrade to idle but record the # attempt so audit shows the admin didn't get the # office they asked for. return_target = "idle" # Update asset with return details if request.usage_company: asset.usage_company = request.usage_company # Same managed-location gating as ``claim_asset``: the 退库 # modal can carry a HRIS-derived ``work_location_name`` that # doesn't match any master row. Route through the resolver # so the row keeps a canonical value (or stays unchanged). location_resolver = await self._get_location_resolver() resolved_location = location_resolver.resolve(request.location) if resolved_location is not None: asset.location = resolved_location if request.admin_id: asset.admin_id = request.admin_id # Store extra fields in operation detail detail = {} if request.return_date: detail["return_date"] = request.return_date.isoformat() else: detail["return_date"] = date.today().isoformat() if request.admin_id: detail["admin_id"] = str(request.admin_id) if request.usage_company: detail["usage_company"] = request.usage_company if request.usage_department: detail["usage_department"] = request.usage_department if resolved_location is not None: detail["location"] = resolved_location if request.location and resolved_location != request.location: detail["location_raw"] = request.location if return_target == "office" and resolved_office is not None: detail["return_target"] = "office" detail["target_office_id"] = str(resolved_office.id) detail["target_office_resolution"] = ( "explicit" if request.target_office_id else "home_office_inferred" ) operation = AssetOperation( operation_number=_generate_operation_number(), asset_id=asset.id, operation_type="return", status="confirmed", from_user_id=from_user_id, operator_id=operator_id, notes=request.notes or "Asset returned to stock", detail=detail if detail else None, confirmed_at=datetime.now(UTC), confirmed_by=operator_id, ) self.db.add(operation) history = AssetHistory( asset_id=asset.id, action="unassigned", from_user=from_user_id, operator_id=operator_id, notes=request.notes or "Asset returned", ) self.db.add(history) asset.assigned_to = None asset.assigned_at = None # Defensive cleanup: clear expected_return_date in case it was # set via admin asset-update path on a claimed asset. The field # is only semantically meaningful while assigned to a holder. asset.expected_return_date = None if return_target == "office" and resolved_office is not None: asset.status = ASSET_STATUS_HELD_BY_OFFICE asset.holder_kind = ASSET_HOLDER_KIND_OFFICE asset.holder_office_id = resolved_office.id # Mirror ``recycle_to_office``: when the admin didn't # explicitly type a new location, sync ``location`` to the # office's city so inventory group-bys reflect where the # device actually lives. ``office.city`` is operator- # managed (asset_office_addresses settings page) and is # NOT guaranteed to be in ``asset_locations.name`` — route # it through the resolver so a misspelled office city # can't reintroduce a dirty value. if not request.location: office_resolved = location_resolver.resolve( resolved_office.city, ) if office_resolved is not None: asset.location = office_resolved else: asset.status = "idle" asset.holder_kind = ASSET_HOLDER_KIND_PERSONAL asset.holder_office_id = None asset.updated_at = datetime.now(UTC) await self.db.flush() return asset, operation async def borrow_asset( self, asset_id: UUID, request: AssetBorrowRequest, operator_id: UUID, ) -> tuple[AssetItem, AssetOperation]: """Temporarily lend an asset. Pre-launch hardening 2026-05-04: borrow no longer overwrites ``assigned_to``. The original holder (if any) is preserved while the asset is on loan, so offboarding scans, manager visibility, and audit trails remain accurate. The borrower is tracked in ``borrowed_by``. ``borrow_return`` clears ``borrowed_by`` and restores the prior status (``claimed`` if there was an original owner, otherwise ``idle``). """ result = await self.db.execute( select(AssetItem) .where(AssetItem.id == asset_id, AssetItem.deleted_at.is_(None)) .with_for_update() ) asset = result.scalar_one_or_none() if not asset: raise NotFoundError("Asset", str(asset_id)) # Borrow can start from an unowned asset (idle/available) OR a # claimed/assigned asset (a colleague lending out their gear) — # in both cases ``assigned_to`` is preserved. if asset.status not in ("idle", "available", "claimed", "assigned"): raise AppException( code="ASSET_NOT_AVAILABLE", message=f"Asset is {asset.status}, cannot borrow", status_code=400, ) if asset.borrowed_by is not None: raise AppException( code="ASSET_ALREADY_BORROWED", message="Asset is already on loan; return it first.", status_code=400, ) user = await self._verify_user(request.to_user_id) # Borrow doesn't change ownership — keep the reservation when # the reserved person themselves borrows it early. self._ensure_not_reserved_for_other( asset, request.to_user_id, clear_on_match=False, ) # ``assigned_to`` is INTENTIONALLY preserved — borrowing does # not change ownership. Only the borrowed_by + status fields # move. now = datetime.now(UTC) asset.borrowed_by = request.to_user_id asset.borrow_started_at = now asset.status = "borrowed" asset.updated_at = now # Update optional fields from request. When not provided, sync from borrower contract entity. resolved_usage_company = request.usage_company or user.company_name if resolved_usage_company: asset.usage_company = resolved_usage_company # Managed-location gating: same contract as claim/return. location_resolver = await self._get_location_resolver() resolved_location = location_resolver.resolve(request.location) if resolved_location is not None: asset.location = resolved_location if request.expected_return_date: asset.expected_return_date = request.expected_return_date # Store extra fields in operation detail detail = {} if request.borrow_date: detail["borrow_date"] = request.borrow_date.isoformat() else: detail["borrow_date"] = date.today().isoformat() if request.expected_return_date: detail["expected_return_date"] = request.expected_return_date.isoformat() # Don't fake today as the expected_return_date — the schema # validator requires a future date, so storing today silently # would contradict the field's contract. Leave the key absent # if the caller didn't supply one; UI renders "未指定". if resolved_usage_company: detail["usage_company"] = resolved_usage_company if request.usage_department: detail["usage_department"] = request.usage_department if resolved_location is not None: detail["location"] = resolved_location if request.location and resolved_location != request.location: detail["location_raw"] = request.location operation = AssetOperation( operation_number=_generate_operation_number(), asset_id=asset.id, operation_type="borrow", status="confirmed", to_user_id=request.to_user_id, operator_id=operator_id, notes=request.notes or f"Borrowed by {user.display_name or user.email}", detail=detail if detail else None, confirmed_at=datetime.now(UTC), confirmed_by=operator_id, ) self.db.add(operation) history = AssetHistory( asset_id=asset.id, action="assigned", to_user=request.to_user_id, operator_id=operator_id, notes=request.notes or f"Borrowed by {user.display_name or user.email}", ) self.db.add(history) await self.db.flush() return asset, operation async def return_borrowed_asset( self, asset_id: UUID, request: AssetBorrowReturnRequest, operator_id: UUID, ) -> tuple[AssetItem, AssetOperation]: """Return an asset from borrow. Restores the original holder if there was one (``status='claimed'`` and ``assigned_to`` is preserved). Otherwise the asset returns to ``idle`` stock. """ result = await self.db.execute( select(AssetItem) .where(AssetItem.id == asset_id, AssetItem.deleted_at.is_(None)) .with_for_update() ) asset = result.scalar_one_or_none() if not asset: from app.core.exceptions import NotFoundError raise NotFoundError("Asset", str(asset_id)) if asset.status != "borrowed": raise AppException( code="ASSET_NOT_BORROWED", message=f"Asset is {asset.status}, not borrowed", status_code=400, ) # The borrower (subject of the operation row) lives in # ``borrowed_by`` post-115. Pre-115 rows are recovered by the # migration backfill, but a defensive fallback to assigned_to # keeps the audit pair sensible if the backfill missed an edge # case. borrower_id = asset.borrowed_by or asset.assigned_to # The original owner (if any) — survives the borrow round-trip. original_owner_id = asset.assigned_to if asset.borrowed_by else None operation = AssetOperation( operation_number=_generate_operation_number(), asset_id=asset.id, operation_type="borrow_return", status="confirmed", from_user_id=borrower_id, to_user_id=original_owner_id, operator_id=operator_id, notes=request.notes or "Borrowed asset returned", confirmed_at=datetime.now(UTC), confirmed_by=operator_id, ) self.db.add(operation) history = AssetHistory( asset_id=asset.id, action="unassigned", from_user=borrower_id, to_user=original_owner_id, operator_id=operator_id, notes=request.notes or "Borrowed asset returned", ) self.db.add(history) asset.borrowed_by = None asset.borrow_started_at = None # Clear borrow-only metadata so a re-borrow doesn't inherit # the previous cycle's expected_return_date as a stale ghost. asset.expected_return_date = None # Status restoration: keep original owner if there was one. if original_owner_id is not None: asset.status = "claimed" else: asset.assigned_to = None asset.assigned_at = None asset.status = "idle" asset.updated_at = datetime.now(UTC) await self.db.flush() return asset, operation async def extend_borrow( self, asset_id: UUID, *, new_expected_return_date: date, reason: str, operator_id: UUID, allow_earlier: bool, trace_id: str = "", ) -> tuple[AssetItem, AssetOperation]: """Extend (or adjust) a borrowed asset's expected return date. No approval — the date moves immediately. ``allow_earlier`` is ``True`` for asset admins (they may also bring the date forward); employee self-service passes ``False`` so a borrower can only push the return date later, never quietly shorten it. Notifications fan out from here (same transaction, outbox-first): * the borrower — a confirmation when they extended it themselves, an FYI when an admin adjusted it; * the borrower's direct lead — always (if active + reachable); * the asset admins — only on employee self-service (an admin who performed the change doesn't notify the whole admin pool). This pairs with the daily ``cron.asset.borrow_overdue`` reaper: pushing the date forward naturally stops the overdue escalation without any extra wiring. """ result = await self.db.execute( select(AssetItem) .where(AssetItem.id == asset_id, AssetItem.deleted_at.is_(None)) .with_for_update() ) asset = result.scalar_one_or_none() if not asset: raise NotFoundError("Asset", str(asset_id)) if asset.status != "borrowed": raise AppException( code="ASSET_NOT_BORROWED", message=f"Asset is {asset.status}, not borrowed", status_code=400, ) if new_expected_return_date <= date.today(): raise AppException( code="INVALID_RETURN_DATE", message="延长日期必须是将来的日期", status_code=400, ) previous = asset.expected_return_date if ( not allow_earlier and previous is not None and new_expected_return_date <= previous ): raise AppException( code="INVALID_RETURN_DATE", message="延长日期必须晚于当前预计归还日期", status_code=400, ) now = datetime.now(UTC) asset.expected_return_date = new_expected_return_date asset.updated_at = now detail = { "reason": reason, "previous_expected_return_date": ( previous.isoformat() if previous else None ), "expected_return_date": new_expected_return_date.isoformat(), # The actor relationship drives the notification fan-out below; # persisting it keeps the audit row self-explanatory. "actor_kind": ( "borrower" if asset.borrowed_by is not None and operator_id == asset.borrowed_by else "admin" ), } operation = AssetOperation( operation_number=_generate_operation_number(), asset_id=asset.id, operation_type="borrow_extension", status="confirmed", to_user_id=asset.borrowed_by, operator_id=operator_id, notes=reason, detail=detail, confirmed_at=now, confirmed_by=operator_id, ) self.db.add(operation) history = AssetHistory( asset_id=asset.id, action="borrow_extended", to_user=asset.borrowed_by, operator_id=operator_id, notes=reason, ) self.db.add(history) await self.db.flush() await self._notify_borrow_extension( asset=asset, previous=previous, new_date=new_expected_return_date, reason=reason, operator_id=operator_id, trace_id=trace_id, ) return asset, operation async def _notify_borrow_extension( self, *, asset: AssetItem, previous: date | None, new_date: date, reason: str, operator_id: UUID, trace_id: str, ) -> None: """Fan-out for :meth:`extend_borrow`. Lazy-imports the notification stack so the heavy ``app.notifications`` graph never loads at ``AssetService`` import time.""" from app.identity.person_models import IdentityPerson # noqa: PLC0415 from app.notifications.service import ( # noqa: PLC0415 NotificationEnvelope, NotificationService, ) from app.scheduled_jobs.failure_notifier import ( # noqa: PLC0415 list_admin_recipients, ) if asset.borrowed_by is None: return borrower = await self.db.get(IdentityPerson, asset.borrowed_by) if borrower is None: return actor_is_borrower = operator_id == asset.borrowed_by asset_label = asset.asset_name or asset.model or asset.asset_tag borrower_name = ( borrower.display_name or (borrower.work_email.split("@", 1)[0] if borrower.work_email else "") ) new_str = new_date.strftime("%Y-%m-%d") prev_str = previous.strftime("%Y-%m-%d") if previous else "未指定" # Build the recipient list, deduped by person id. # Tuple = (person, role) where role ∈ {borrower, manager, admin}. recipients: list[tuple[IdentityPerson, str]] = [] seen: set[UUID] = set() if borrower.work_email and borrower.deleted_at is None: recipients.append((borrower, "borrower")) seen.add(borrower.id) if borrower.manager_person_id is not None: manager = await self.db.get(IdentityPerson, borrower.manager_person_id) if ( manager is not None and manager.id not in seen and manager.id != operator_id and manager.deleted_at is None and manager.employment_status == "active" and manager.work_email ): recipients.append((manager, "manager")) seen.add(manager.id) # Admins are notified only when the borrower extended it themselves # (self-service). An admin acting on the asset is already aware and # we don't fan out to the whole admin pool for their own action. if actor_is_borrower: for admin in await list_admin_recipients(self.db): if admin.id not in seen and admin.id != operator_id: recipients.append((admin, "admin")) seen.add(admin.id) if not recipients: return notif = NotificationService(self.db) for person, role in recipients: if role == "borrower": if actor_is_borrower: title = f"[借用延期] 已延长至 {new_str}" body = ( f"你已将借用的『{asset_label}』预计归还时间延长至 {new_str}" f"(原 {prev_str})。\n延期原因:{reason}" ) else: title = f"[借用延期] 预计归还时间已调整为 {new_str}" body = ( f"管理员已将你借用的『{asset_label}』预计归还时间调整为 " f"{new_str}(原 {prev_str})。\n说明:{reason}" ) else: title = f"[借用延期] {borrower_name} 延长至 {new_str}" body = ( f"{borrower_name} 因『{reason}』将借用的『{asset_label}』" f"预计归还时间延长至 {new_str}(原 {prev_str})。" ) env = NotificationEnvelope( recipient=person, title=title, body_text=body, channels=["feishu", "email"], template_key=( "asset.borrow_extension.borrower" if role == "borrower" else "asset.borrow_extension.stakeholder" ), context={ "asset_tag": asset.asset_tag, "asset_label": asset_label, "borrower_name": borrower_name, "expected_return_date": new_str, "previous_expected_return_date": prev_str, "reason": reason, "recipient_role": role, }, related_resource_type="asset", related_resource_id=asset.id, ) try: await notif.dispatch( env, trace_id=trace_id, actor_user_id=operator_id, ) except Exception: # noqa: BLE001 # Outbox row already persisted; the retry worker recovers. # A notification hiccup must not roll back the extension. logger.exception( "failed to dispatch borrow-extension notification to %s", person.id, ) async def change_assignee( self, asset_id: UUID, request: AssetChangeAssigneeRequest, operator_id: UUID, ) -> tuple[AssetItem, AssetOperation]: """Change who holds the asset (transfer).""" # Row lock — two concurrent 变更领用人 on the same asset would # otherwise both read the same assigned_to, both write, last # writer wins silently (lost update). result = await self.db.execute( select(AssetItem).where(AssetItem.id == asset_id, AssetItem.deleted_at.is_(None)).with_for_update() ) asset = result.scalar_one_or_none() if asset is None: raise NotFoundError("Asset", str(asset_id)) # Pre-launch hardening 2026-05-04: REJECT borrowed assets. # Previously this method accepted ``borrowed`` and rewrote # ``assigned_to`` — which silently transferred ownership to # whoever currently held the loan. Force a borrow_return first # so the original owner is preserved and the audit pair is # explicit. if asset.status not in ("claimed", "assigned"): raise AppException( code="ASSET_NOT_HELD", message=( f"Asset is {asset.status}; cannot change assignee. " "If on loan, return it first." ), status_code=400, ) if asset.assigned_to == request.to_user_id: raise AppException( code="SAME_USER", message="Asset is already assigned to this user", status_code=400, ) new_user = await self._verify_user(request.to_user_id) self._ensure_not_reserved_for_other(asset, request.to_user_id) from_user_id = asset.assigned_to # Update asset with new assignee details asset.assigned_to = request.to_user_id asset.assigned_at = datetime.now(UTC) asset.updated_at = datetime.now(UTC) # Update optional fields from request. When not provided, sync from new assignee contract entity. resolved_usage_company = request.usage_company or new_user.company_name if resolved_usage_company: asset.usage_company = resolved_usage_company # Managed-location gating: same contract as claim/return/borrow. location_resolver = await self._get_location_resolver() resolved_location = location_resolver.resolve(request.location) if resolved_location is not None: asset.location = resolved_location # Store extra fields in operation detail detail = {} if request.claim_date: detail["claim_date"] = request.claim_date.isoformat() else: detail["claim_date"] = date.today().isoformat() if resolved_usage_company: detail["usage_company"] = resolved_usage_company if request.usage_department: detail["usage_department"] = request.usage_department if resolved_location is not None: detail["location"] = resolved_location if request.location and resolved_location != request.location: detail["location_raw"] = request.location operation = AssetOperation( operation_number=_generate_operation_number(), asset_id=asset.id, operation_type="transfer", status="confirmed", from_user_id=from_user_id, to_user_id=request.to_user_id, operator_id=operator_id, notes=request.notes or f"Transferred to {new_user.display_name or new_user.email}", detail=detail if detail else None, confirmed_at=datetime.now(UTC), confirmed_by=operator_id, ) self.db.add(operation) history = AssetHistory( asset_id=asset.id, action="transferred", from_user=from_user_id, to_user=request.to_user_id, operator_id=operator_id, notes=request.notes or f"Transferred to {new_user.display_name or new_user.email}", ) self.db.add(history) await self.db.flush() return asset, operation async def repair_asset( self, asset_id: UUID, request: AssetRepairRequest, operator_id: UUID, ) -> tuple[AssetItem, AssetOperation]: """Send an asset to repair.""" # Row lock — prevent repair from racing with a concurrent # dispose/return on the same asset. result = await self.db.execute( select(AssetItem).where(AssetItem.id == asset_id, AssetItem.deleted_at.is_(None)).with_for_update() ) asset = result.scalar_one_or_none() if asset is None: raise NotFoundError("Asset", str(asset_id)) if asset.status in ("disposed", "disposal_pending"): raise AppException( code="ASSET_DISPOSED", message=f"Asset is {asset.status}, cannot repair", status_code=400, ) # Round-21 review fix-up 2026-05-12: repair was previously a # denylist (anything not disposed/retired). A ``held_by_leaver`` # asset is physically NOT in IT custody, so sending it to # repair is nonsense and — worse — silently rewrites status to # ``maintenance``, then ``repair_complete`` rewrites to ``idle``, # losing the ex-employee attribution. Reject explicitly. if asset.status == ASSET_STATUS_HELD_BY_LEAVER: raise AppException( code="ASSET_HELD_BY_LEAVER", message=( "Asset is held by a departed employee; recover it " "first (asset detail page → 追回) before sending to " "repair." ), status_code=400, ) from_user_id = asset.assigned_to operation = AssetOperation( operation_number=_generate_operation_number(), asset_id=asset.id, operation_type="repair", status="confirmed", from_user_id=from_user_id, operator_id=operator_id, notes=request.notes or "Sent to repair", detail=request.detail, confirmed_at=datetime.now(UTC), confirmed_by=operator_id, ) self.db.add(operation) asset.device_status = "under_repair" asset.updated_at = datetime.now(UTC) history = AssetHistory( asset_id=asset.id, action="updated", operator_id=operator_id, notes=request.notes or "Sent to repair", ) self.db.add(history) await self.db.flush() return asset, operation async def complete_repair( self, asset_id: UUID, request: AssetRepairCompleteRequest, operator_id: UUID, ) -> tuple[AssetItem, AssetOperation]: """Mark repair as done.""" # Row lock — repair-complete is a narrow state flip but still # races with dispose/return. Match the other lifecycle writers. result = await self.db.execute( select(AssetItem).where(AssetItem.id == asset_id, AssetItem.deleted_at.is_(None)).with_for_update() ) asset = result.scalar_one_or_none() if asset is None: raise NotFoundError("Asset", str(asset_id)) if asset.device_status != "under_repair": raise AppException( code="ASSET_NOT_UNDER_REPAIR", message=f"Asset device_status is {asset.device_status}, not under_repair", status_code=400, ) operation = AssetOperation( operation_number=_generate_operation_number(), asset_id=asset.id, operation_type="repair_complete", status="confirmed", operator_id=operator_id, notes=request.notes or "Repair completed", detail=request.detail, confirmed_at=datetime.now(UTC), confirmed_by=operator_id, ) self.db.add(operation) asset.device_status = "normal" asset.updated_at = datetime.now(UTC) history = AssetHistory( asset_id=asset.id, action="updated", operator_id=operator_id, notes=request.notes or "Repair completed", ) self.db.add(history) await self.db.flush() return asset, operation async def bulk_intake_assets( self, upload: UploadFile, operator_id: UUID, ) -> BulkAssetMutationResponse: rows = await read_bulk_rows(upload) results: list[BulkAssetRowResult] = [] successful_rows = 0 for index, row in enumerate(rows, start=2): asset_tag = row.get("asset_tag") try: async with self.db.begin_nested(): payload = _normalize_bulk_row(row) # Round 2b audit fix-up 2026-05-04: Round 1 removed # ``assigned_to`` from ``AssetIntakeRequest``, but # legacy CSVs from prior template revisions still # carry the column. ``extra='ignore'`` (Pydantic # default) would silently drop it, leaving the # admin thinking the upload assigned the asset # when it actually landed idle. Pop the field and # claim explicitly after intake. pending_assignee = payload.pop("assigned_to", None) request = AssetIntakeRequest(**payload) asset, _operation = await self.intake_asset(request, operator_id) if pending_assignee is not None: try: await self.claim_asset( asset.id, AssetClaimRequest(to_user_id=pending_assignee), operator_id, ) except ( AppException, ValidationError, PydanticValidationError, ) as claim_exc: # Surface the claim failure to the caller — # the row succeeds the intake but the # message tells the admin to claim manually. # ``PydanticValidationError`` is named here # because a pydantic model IS constructed inside # this block (``AssetClaimRequest``). Until the # aliased import above, the bare # ``ValidationError`` resolved to the Helix one, # so pydantic would have escaped this handler, # aborted the row's savepoint, and thrown away an # otherwise successful intake as a plain "failed" # row. Today ``_normalize_bulk_row`` pre-coerces # ``assigned_to`` to a ``UUID`` (a bad cell fails # earlier, as a "failed" row, by design), so this # is a guard against a future field/validator # rather than a live escape — but the guard now # actually catches what it names. The two sibling # handlers below only appeared correct because # they list ``ValueError``, which pydantic's # ``ValidationError`` happens to subclass. results.append( BulkAssetRowResult( row_number=index, asset_tag=request.asset_tag, asset_id=asset.id, status="created_unclaimed", message=( f"intake OK, claim to " f"{pending_assignee} failed: " f"{claim_exc}" ), ) ) successful_rows += 1 continue successful_rows += 1 results.append( BulkAssetRowResult( row_number=index, asset_tag=request.asset_tag, asset_id=asset.id, status="created", message="created", ) ) except ( AppException, ValidationError, PydanticValidationError, ValueError, ) as exc: results.append( BulkAssetRowResult( row_number=index, asset_tag=asset_tag, status="failed", message=str(exc), ) ) return BulkAssetMutationResponse( total_rows=len(rows), successful_rows=successful_rows, failed_rows=len(rows) - successful_rows, results=results, ) async def bulk_update_assets( self, upload: UploadFile, operator_id: UUID, ) -> BulkAssetMutationResponse: rows = await read_bulk_rows(upload) results: list[BulkAssetRowResult] = [] successful_rows = 0 for index, row in enumerate(rows, start=2): asset_tag = row.get("asset_tag") try: async with self.db.begin_nested(): payload = _normalize_bulk_row(row) asset_tag = payload.pop("asset_tag", asset_tag) if not asset_tag: raise AppException( code="ASSET_TAG_REQUIRED", message="asset_tag is required for bulk update", status_code=400, ) query = await self.db.execute( select(AssetItem).where(AssetItem.asset_tag == asset_tag) ) asset = query.scalar_one_or_none() if not asset: raise NotFoundError("Asset", str(asset_tag)) request = AssetUpdateRequest(**payload) if not request.model_dump(exclude_unset=True): raise AppException( code="NO_UPDATE_FIELDS", message="No updatable fields provided", status_code=400, ) updated = await self.update_asset(asset.id, request, operator_id) successful_rows += 1 results.append( BulkAssetRowResult( row_number=index, asset_tag=str(asset_tag), asset_id=updated.id, status="updated", message="updated", ) ) except ( AppException, ValidationError, PydanticValidationError, ValueError, ) as exc: results.append( BulkAssetRowResult( row_number=index, asset_tag=str(asset_tag) if asset_tag else None, status="failed", message=str(exc), ) ) return BulkAssetMutationResponse( total_rows=len(rows), successful_rows=successful_rows, failed_rows=len(rows) - successful_rows, results=results, ) async def _check_bulk_idempotency( self, action_type: str, idempotency_key: str | None, operator_id: "UUID | None" = None, ) -> bool: """Check if a bulk operation with this idempotency_key already executed. Scoped by operator_id so two different admins using the same client-generated key don't falsely collide — each admin's key lives in its own namespace. """ if not idempotency_key: return False filters = [ AssetOperation.operation_type == action_type, AssetOperation.detail["idempotency_key"].as_string() == idempotency_key, ] if operator_id is not None: filters.append(AssetOperation.operator_id == operator_id) result = await self.db.execute( select(func.count()).select_from(AssetOperation).where(and_(*filters)) ) return (result.scalar() or 0) > 0 async def _acquire_bulk_idempotency_lock( self, action_type: str, idempotency_key: str | None, operator_id: "UUID | None" = None, ) -> bool: """Reserve ``idempotency_key`` for the bulk operation about to run. Two-step idempotency contract — call this BEFORE iterating rows and BEFORE :meth:`_check_bulk_idempotency`. The Redis SETNX gives us a short-lived per-key serialization point so two concurrent requests carrying the same idempotency key cannot both pass the DB-side check at line 1569 (which is purely advisory and races with the post-loop stamp at line 1610). Returns True if this caller wins the lock, False if another caller is already in-flight or has finished — both cases must surface to the client as ``DUPLICATE_BULK_OPERATION``. TTL is set to 15 minutes — generous enough that a slow batch (~hundreds of assets, per-row Excel writes) finishes before the lock auto-expires, short enough that an orphaned lock from a crashed worker doesn't permanently block the key. Scoped by operator_id so two different admins sharing the same client-generated idempotency key don't block each other — each admin's lock lives in its own namespace. **Fail-closed** on Redis unavailability: raises ``AppException`` 503 instead of letting the request through with the weaker DB-only check. Bulk operations are admin-initiated and re-tryable; a minute-scale Redis outage causing a 503 is recoverable (admin retries when Redis is back), but a silent double-bulk would mutate ``asset_items`` rows twice and is not recoverable without manual reconciliation. Production-grade choice: never mutate state on a degraded path. """ if not idempotency_key: return True # Import + client construction are NOT Redis I/O — keep them OUTSIDE # the try so a programming error (wrong module path, bad config) # propagates as a real 500 instead of being mislabelled as a # transient "Redis outage" 503. A typo here (``app.core.redis`` vs # ``app.core.redis_client``) silently turned every bulk op into a # 503 in prod until the backend logs were read — the broad except # below must only ever guard the network call. from app.core.redis_client import get_redis redis = get_redis() # Scope by operator_id so different admins with the same # client-side key don't collide with each other. scope = str(operator_id) if operator_id else "anon" key = f"asset:bulk:{action_type}:{scope}:{idempotency_key}" try: # SET NX with 15-minute TTL: SETNX-equivalent atomic reserve. won = await redis.set(key, "1", nx=True, ex=900) return bool(won) except Exception as exc: # noqa: BLE001 logger.error( "bulk idempotency lock unavailable (Redis error: %s) — " "rejecting request to avoid silent double-bulk; admin " "should retry once Redis is healthy", exc, ) raise AppException( code="IDEMPOTENCY_LOCK_UNAVAILABLE", message=( "Idempotency lock backend unavailable; refusing " "bulk operation to prevent silent duplicate writes. " "Retry once the platform is healthy." ), status_code=503, ) from exc async def bulk_claim_assets( self, request: BulkClaimRequest, operator_id: UUID, ) -> BulkAssetMutationResponse: """Batch claim/dispatch multiple assets to the same user.""" if not await self._acquire_bulk_idempotency_lock( "claim", request.idempotency_key, operator_id, ): raise AppException( code="DUPLICATE_BULK_OPERATION", message=f"Bulk claim with idempotency_key={request.idempotency_key} already in flight or completed", status_code=409, ) if await self._check_bulk_idempotency("claim", request.idempotency_key, operator_id): raise AppException( code="DUPLICATE_BULK_OPERATION", message=f"Bulk claim with idempotency_key={request.idempotency_key} already executed", status_code=409, ) await self._verify_user(request.to_user_id) results: list[BulkAssetRowResult] = [] successful = 0 for index, asset_id in enumerate(request.asset_ids, start=1): try: async with self.db.begin_nested(): claim_req = AssetClaimRequest( to_user_id=request.to_user_id, claim_date=request.claim_date, usage_company=request.usage_company, usage_department=request.usage_department, location=request.location, notes=request.notes, ) asset, op = await self.claim_asset(asset_id, claim_req, operator_id) # Tag operation with idempotency_key. REASSIGN the dict # rather than mutating it in place: ``AssetOperation.detail`` # is a plain ``JSON`` column (no ``MutableDict`` tracking), # and ``claim_asset`` already populated + flushed ``detail``, # so an in-place ``op.detail[k] = v`` is invisible to # SQLAlchemy's change detection and never persisted — which # silently defeated the DB-side idempotency guard # (``_check_bulk_idempotency`` read back nothing once the # 15-min Redis lock expired, re-running the whole batch). if request.idempotency_key: op.detail = { **(op.detail or {}), "idempotency_key": request.idempotency_key, } successful += 1 results.append(BulkAssetRowResult( row_number=index, asset_tag=asset.asset_tag, asset_id=asset.id, status="claimed", operation_number=op.operation_number, )) except (AppException, ValueError) as exc: results.append(BulkAssetRowResult( row_number=index, asset_id=asset_id, status="failed", message=str(exc), )) return BulkAssetMutationResponse( total_rows=len(request.asset_ids), successful_rows=successful, failed_rows=len(request.asset_ids) - successful, results=results, ) async def bulk_borrow_assets( self, request: BulkBorrowRequest, operator_id: UUID, ) -> BulkAssetMutationResponse: """Batch borrow multiple assets to the same user.""" if not await self._acquire_bulk_idempotency_lock( "borrow", request.idempotency_key, operator_id, ): raise AppException( code="DUPLICATE_BULK_OPERATION", message=f"Bulk borrow with idempotency_key={request.idempotency_key} already in flight or completed", status_code=409, ) if await self._check_bulk_idempotency("borrow", request.idempotency_key, operator_id): raise AppException( code="DUPLICATE_BULK_OPERATION", message=f"Bulk borrow with idempotency_key={request.idempotency_key} already executed", status_code=409, ) await self._verify_user(request.to_user_id) results: list[BulkAssetRowResult] = [] successful = 0 for index, asset_id in enumerate(request.asset_ids, start=1): try: async with self.db.begin_nested(): borrow_req = AssetBorrowRequest( to_user_id=request.to_user_id, borrow_date=request.borrow_date, expected_return_date=request.expected_return_date, usage_company=request.usage_company, usage_department=request.usage_department, location=request.location, notes=request.notes, ) asset, op = await self.borrow_asset(asset_id, borrow_req, operator_id) # Tag operation with idempotency_key — reassign (not in-place # mutate) so the plain JSON column change is persisted. See # the claim path above for the full rationale. if request.idempotency_key: op.detail = { **(op.detail or {}), "idempotency_key": request.idempotency_key, } successful += 1 results.append(BulkAssetRowResult( row_number=index, asset_tag=asset.asset_tag, asset_id=asset.id, status="borrowed", operation_number=op.operation_number, )) except (AppException, ValueError) as exc: results.append(BulkAssetRowResult( row_number=index, asset_id=asset_id, status="failed", message=str(exc), )) return BulkAssetMutationResponse( total_rows=len(request.asset_ids), successful_rows=successful, failed_rows=len(request.asset_ids) - successful, results=results, ) async def dispose_asset( self, asset_id: UUID, request: AssetDisposeRequest, operator_id: UUID, *, force: bool = False, allow_held_by_leaver: bool = False, allow_direct_held_disposal: bool = False, disposal_context: str | None = None, ) -> tuple[AssetItem, AssetOperation]: """Dispose/retire an asset. ``force=True`` bypasses the "asset is in use" guard and is used by the auto-execute approval callbacks — specifically offboarding buyback, where the leaver literally holds the asset at the moment we mark it sold. The assigned_to reference is still cleared below as a side effect, so the asset ends up in a clean disposed state. The already-disposed guard always applies. ``allow_held_by_leaver`` is the in-process escape hatch for the same buyback path — moved off the Pydantic schema so an external caller can't sneak it in via a request body. The HTTP router always invokes with the default ``False``; trusted in-process callers (buyback auto-execute) set it explicitly. ``disposal_context`` names WHY the disposal happened when the ``disposal_type`` alone doesn't say it — currently only ``"buyback"``, passed by the offboarding-buyback approval callback. It reaches the MDM cleanup card so an administrator reads "离职回购(员工买断)" instead of a bare "sold". Stated explicitly rather than inferred from ``allow_held_by_leaver``: that flag happens to be buyback-only today, and a cleanup card that lies about why a machine left is worse than one that says less. """ result = await self.db.execute( select(AssetItem).where(AssetItem.id == asset_id, AssetItem.deleted_at.is_(None)).with_for_update() ) asset = result.scalar_one_or_none() if not asset: raise NotFoundError("Asset", str(asset_id)) if asset.status in ("disposed", "disposal_pending"): raise AppException( code="ASSET_ALREADY_DISPOSED", message=f"Asset is already {asset.status}", status_code=400, ) direct_held_disposal_types = ("sold", "lost") if ( asset.status in ("assigned", "claimed", "borrowed") and not ( allow_direct_held_disposal and request.disposal_type in direct_held_disposal_types ) and not force ): raise AppException( code="ASSET_IN_USE", message=( f"Asset is {asset.status}; return it first before " f"{request.disposal_type} disposal" ), status_code=400, ) # held_by_leaver guard — only fires when force=True. The # admin "Lost / write-off" path (force=False, default) is # the legitimate way to dispose of a held_by_leaver asset: # the asset is no longer "assigned"/"claimed"/"borrowed" so # the in-use guard already lets it through, and the audit # row's from_user_id preserves the prior leaver attribution. # The force=True branch, by contrast, is only meant for the # buyback auto-execute callback (which sets # allow_held_by_leaver=True explicitly). Any OTHER force=True # caller hitting a held_by_leaver row would silently bypass # both the in-use guard AND the leaver attribution check — # reject loudly unless the caller opts in. if ( asset.status == ASSET_STATUS_HELD_BY_LEAVER and force and not allow_held_by_leaver ): raise AppException( code="ASSET_HELD_BY_LEAVER", message=( "Asset is held_by_leaver. Use the standard " "dispose path (without force=True), run " "/assets/{id}/recover-from-leaver first, or set " "allow_held_by_leaver=True on the dispose request " "if this is the buyback auto-execute path." ), status_code=409, ) from_user_id = asset.assigned_to # Disposal evidence (回收单 / 二手买家付款截图 / 租赁归还回执 / # 硬盘销毁照片). Re-sanitised even though ``AssetDisposeRequest`` # already validated it: a model can be built with # ``model_construct`` or have the field reassigned after # validation, and this method is the last writer before the keys # reach the DB — and later ``presign_get`` verbatim. Same # defence-in-depth stance as ``inventory_service.respond``. # Read as a plain attribute, not ``getattr(..., None)``: every # caller passes a real request model, so a future caller that # doesn't should fail loudly rather than silently drop photos. try: disposal_photos = sanitize_disposal_photo_keys( request.disposal_photos, ) except ValueError as exc: raise AppException( code="ASSET_DISPOSAL_PHOTOS_INVALID", message=str(exc), status_code=422, ) from exc if disposal_photos: # Claims the durable reference on each upload grant (so the # orphan sweep leaves them alone) AND verifies the landed # object matches the size / type / checksum the operator # declared at presign time. Fails the whole disposal on # mismatch — a disposal that silently drops its proof is # worse than one the operator has to retry. from app.core.storage_uploads import mark_uploads_referenced await mark_uploads_referenced( self.db, object_keys=disposal_photos, person_id=operator_id, allowed_purposes=frozenset({"asset_disposal_evidence"}), resource_type="asset", resource_id=str(asset.id), ) # Freeze the disposal financials (net book value AT disposal, # proceeds, gain/loss) from the depreciation policy at this # instant — reproducible even if the policy changes later. from app.domains.asset import financial_service as _fin disposed_on = datetime.now(UTC).date() fin = await _fin.compute_disposal_financials( self.db, asset, request.disposal_proceeds, disposed_on, request.disposal_currency, ) disposal_book_value = fin.pop("_book_value", None) # Pre-launch P1 fix (Round 45, CR-414): snapshot the asset's # financial + identity metadata into the disposal operation's # detail JSON. AssetItem soft-deletes are clean-up candidates; # if the row is later cleaned up the source of truth for # "what was this asset worth on disposal" is lost. The audit # row is the canonical immutable trail. disposal_detail: dict[str, Any] = { "disposal_type": request.disposal_type, "disposal_financials": fin, # Copied into the immutable operation detail for the same # reason as the financials below: ``asset_items`` rows are # soft-delete cleanup candidates, the operation row is the # canonical trail. Keys only — presign at render time. "disposal_photos": disposal_photos, "asset_snapshot": { "asset_tag": asset.asset_tag, "asset_name": asset.asset_name, "brand": asset.brand, "model": asset.model, "serial_number": asset.serial_number, "category": asset.category, "purchase_date": ( asset.purchase_date.isoformat() if asset.purchase_date else None ), "purchase_cost": ( str(asset.purchase_cost) if asset.purchase_cost is not None else None ), "purchase_entity": asset.purchase_entity, "prior_status": asset.status, "prior_assigned_to": ( str(asset.assigned_to) if asset.assigned_to else None ), }, } operation = AssetOperation( operation_number=_generate_operation_number(), asset_id=asset.id, operation_type="disposal", status="confirmed", from_user_id=from_user_id, operator_id=operator_id, notes=request.notes or f"Disposed: {request.disposal_type}", detail=disposal_detail, confirmed_at=datetime.now(UTC), confirmed_by=operator_id, ) self.db.add(operation) asset.status = "disposed" asset.disposal_type = request.disposal_type asset.disposed_at = datetime.now(UTC) # Quantise to 2dp so the in-memory value matches the DB column # scale post-commit (expire_on_commit=False keeps the request's # raw Decimal otherwise — "3000" vs the stored "3000.00"). asset.disposal_proceeds = ( request.disposal_proceeds.quantize(Decimal("0.01")) if request.disposal_proceeds is not None else None ) asset.disposal_book_value = disposal_book_value asset.disposal_currency = fin["disposal_currency"] asset.disposal_photos = list(disposal_photos) if disposal_photos else None asset.assigned_to = None asset.assigned_at = None # A disposed device can't be handed to an incoming hire — clear # any earmark so the hardware-prep queue resurfaces the hire as # un-prepped instead of rendering a reserved-but-disposed ghost. asset.reserved_for_person_id = None asset.reserved_at = None # Disposed assets must clear borrow-state too. If the offboarding # buyback path force-disposes a borrowed asset, leaving # ``borrowed_by`` populated would (a) keep a leaver attached to # a status=disposed row, and (b) cause downstream "due back" # reports / residual scans that filter on ``borrowed_by`` to # misattribute the now-retired asset. asset.borrowed_by = None asset.borrow_started_at = None # Disposed assets must not carry an active expected_return_date — # a borrow that never returned and was disposed should not # surface in "due back" reports indefinitely. asset.expected_return_date = None # Clear office-held holder fields (migration 183). A disposed # device no longer belongs to any office; leaving holder_kind= # 'office' would keep it visible in office-admin inventory # cycles for an asset that doesn't exist anymore. asset.holder_kind = ASSET_HOLDER_KIND_PERSONAL asset.holder_office_id = None asset.updated_at = datetime.now(UTC) history = AssetHistory( asset_id=asset.id, action="retired", operator_id=operator_id, notes=request.notes or f"Disposed: {request.disposal_type}", ) self.db.add(history) await self.db.flush() # A disposed device that is still enrolled in an MDM has to come # out of the tenant — this is the ONLY trigger for that, and the # reason the Jamf administrator gave for it: a machine leaving # company ownership is the one case where his cleanup is real # work. Departures deliberately raise nothing. # # Enqueued here rather than in the two callers so every disposal # path is covered by construction (the HTTP route, the buyback # approval callback, and whatever comes next), and inside the # caller's transaction so a committed disposal can never leave an # un-raised cleanup behind — nothing re-scans historical # disposals to catch one up. The lookup is a local indexed read of # the device ledger; no vendor call happens on this path, because # an HTTP round-trip inside an approval callback's open # transaction is how this platform has previously earned dropped # connections. # SAVEPOINT-isolated. The disposal is the user's action and the # cleanup request is a consequence of it: a bug or an integrity # edge in raising the todo must not take down the disposal itself, # least of all on the buyback approval callback, where the failure # would surface as an approved buyback that never marked the asset # sold. ``best_effort`` still re-raises transaction-aborting DB # errors (1205 / 1213), which the caller genuinely cannot continue # past. A swallowed failure leaves the device un-flagged; the next # ``jamf.device_sync`` does not re-raise it, so the warning below # is the trail — see the module docstring's note that nothing # re-scans historical disposals. from app.core.coordinators.mdm_device_retirement_coordinator import ( request_cleanup_for_disposed_asset, ) async with best_effort( self.db, label=f"raise MDM cleanup for disposed asset {asset.asset_tag}", log=logger, ): await request_cleanup_for_disposed_asset( self.db, asset=asset, disposal_type=request.disposal_type, operator_id=operator_id, last_holder_id=from_user_id, via_buyback=(disposal_context == "buyback"), trace_id=f"asset.dispose:{asset.id}", ) return asset, operation # ------------------------------------------------------------------ # Held-by-leaver: assets attributed to an ex-employee who did not # physically return them. The status is set either automatically by # the offboarding console (``skip_blocker`` on a hardware blocker) # or manually by an asset admin on the detail page. ``assigned_to`` # is preserved so "last holder" attribution and audit are intact. # ------------------------------------------------------------------ async def mark_held_by_leaver( self, asset_id: UUID, *, reason: str, operator_id: UUID, source: str = "manual", leaver_person_id: UUID | None = None, ) -> tuple[AssetItem, AssetOperation]: """Mark an asset as held by a departed employee. ``source`` is ``"offboarding"`` when triggered by the offboarding skip-blocker flow, ``"manual"`` when an admin clicks the action on the asset detail page. Two entry shapes: * **status ∈ ASSET_IN_USE_STATUSES** (assigned / claimed / borrowed + pending variants) — original path. The holder is implicit (``asset.assigned_to``); ``leaver_person_id`` is ignored even if passed. ``assigned_to`` / ``borrowed_by`` are preserved so "last holder" attribution is intact. * **status ∈ {idle, available}** (no current holder) — admin retroactively records "this idle laptop is actually still with departed employee X". ``leaver_person_id`` is REQUIRED and MUST resolve to an existing ``identity_persons`` row (deleted_at IS NULL). The service then writes that id into ``assigned_to`` *and* flips status, atomically. Without a Person row we'd fabricate untraceable attribution — that path is rejected on purpose (see DECISIONS.md / memory ``classification_requires_proof_not_pattern``). Disposed / repair / pending transfer states are still rejected. Idempotent: already ``held_by_leaver`` is a no-op. """ from app.identity.person_models import IdentityPerson result = await self.db.execute( select(AssetItem) .where(AssetItem.id == asset_id, AssetItem.deleted_at.is_(None)) .with_for_update() ) asset = result.scalar_one_or_none() if asset is None: raise NotFoundError("Asset", str(asset_id)) if asset.status == ASSET_STATUS_HELD_BY_LEAVER: # Idempotent — surface a no-op operation so callers (the # offboarding skip flow in particular) don't blow up on a # second click. operation = AssetOperation( operation_number=_generate_operation_number(), asset_id=asset.id, operation_type="mark_held_by_leaver", status="confirmed", from_user_id=asset.assigned_to, operator_id=operator_id, notes=f"[noop, already held_by_leaver] {reason}"[:2000], detail={"source": source, "noop": True}, confirmed_at=datetime.now(UTC), confirmed_by=operator_id, ) self.db.add(operation) await self.db.flush() return asset, operation # Two valid sources: a currently-held row, OR a no-holder row # together with an explicit ``leaver_person_id``. Anything else # (disposed, repair, pending workflow) is still rejected. idle_holder_backfill = asset.status in ("idle", "available") if asset.status not in ASSET_IN_USE_STATUSES and not idle_holder_backfill: raise AppException( code="ASSET_NOT_HELD", message=( f"Asset is {asset.status}; only a held / idle asset " "can be marked as held by a departed employee." ), status_code=400, ) if idle_holder_backfill: if leaver_person_id is None: raise AppException( code="LEAVER_PERSON_REQUIRED", message=( "Asset has no current holder; leaver_person_id " "must be supplied to attribute the held-by-leaver " "state to a specific departed employee." ), status_code=400, ) person_check = await self.db.execute( select(IdentityPerson.id).where( IdentityPerson.id == str(leaver_person_id), IdentityPerson.deleted_at.is_(None), ) ) if person_check.scalar_one_or_none() is None: raise AppException( code="LEAVER_PERSON_NOT_FOUND", message=( "leaver_person_id does not resolve to an existing " "person; create the Person row first." ), status_code=400, ) prior_status = asset.status prior_assignee = asset.assigned_to prior_borrower = asset.borrowed_by if idle_holder_backfill: # Stamp the leaver into ``assigned_to`` so the asset detail # page / audit chain has the same shape as a normal # in-use → leaver transition. ``recover_from_leaver`` will # later wipe it back to NULL, matching the prior idle state. asset.assigned_to = leaver_person_id asset.status = ASSET_STATUS_HELD_BY_LEAVER asset.updated_at = datetime.now(UTC) # For the in-use path: ``assigned_to`` and ``borrowed_by`` are # INTENTIONALLY preserved so audit / "last holder" surfaces stay # accurate. The frontend renders both with a red 已离职 chip. # In-use path: leaver is the prior holder (frozen in place) → # log as from_user_id, to_user_id=NULL. # Backfill path: no prior holder, attributing forward → # from_user_id=NULL, to_user_id=leaver. asset_history.from_user # uses the *resulting* attributed leaver in both cases so the # detail-page "last holder" timeline reads cleanly. operation = AssetOperation( operation_number=_generate_operation_number(), asset_id=asset.id, operation_type="mark_held_by_leaver", status="confirmed", from_user_id=None if idle_holder_backfill else prior_assignee, to_user_id=leaver_person_id if idle_holder_backfill else None, operator_id=operator_id, notes=reason[:2000] if reason else None, detail={ "source": source, "prior_status": prior_status, "prior_assigned_to": ( str(prior_assignee) if prior_assignee else None ), "prior_borrowed_by": ( str(prior_borrower) if prior_borrower else None ), # Visibility (review-r2): the incoming-hire earmark is # DELIBERATELY preserved (held_by_leaver is a reservable # state — the inherit_from_leaver flow), but if the # leaver ultimately keeps the device the asset admin # needs the trail to see whose prep plan it breaks. "hire_reservation_pending": ( str(asset.reserved_for_person_id) if asset.reserved_for_person_id else None ), **( {"backfilled_leaver_person_id": str(leaver_person_id)} if idle_holder_backfill else {} ), }, confirmed_at=datetime.now(UTC), confirmed_by=operator_id, ) self.db.add(operation) history = AssetHistory( asset_id=asset.id, action="held_by_leaver", from_user=( leaver_person_id if idle_holder_backfill else prior_assignee ), operator_id=operator_id, notes=reason[:2000] if reason else None, ) self.db.add(history) await self.db.flush() return asset, operation async def recover_from_leaver( self, asset_id: UUID, *, notes: str | None, operator_id: UUID, ) -> tuple[AssetItem, AssetOperation]: """Recover a held-by-leaver asset back to idle stock. Used when the ex-employee finally returns the laptop / hardware. Clears ``assigned_to``, ``assigned_at``, ``borrowed_by``, ``borrow_started_at``, ``expected_return_date`` and flips the status to ``idle`` so the asset can be re-claimed by another person via the normal claim flow. """ result = await self.db.execute( select(AssetItem) .where(AssetItem.id == asset_id, AssetItem.deleted_at.is_(None)) .with_for_update() ) asset = result.scalar_one_or_none() if asset is None: raise NotFoundError("Asset", str(asset_id)) if asset.status != ASSET_STATUS_HELD_BY_LEAVER: raise AppException( code="ASSET_NOT_HELD_BY_LEAVER", message=( f"Asset is {asset.status}; recover_from_leaver is only " "valid for assets currently marked held_by_leaver." ), status_code=400, ) prior_assignee = asset.assigned_to prior_borrower = asset.borrowed_by operation = AssetOperation( operation_number=_generate_operation_number(), asset_id=asset.id, operation_type="recover_from_leaver", status="confirmed", from_user_id=prior_assignee or prior_borrower, operator_id=operator_id, notes=notes[:2000] if notes else "Recovered from ex-employee", detail={ "prior_assigned_to": ( str(prior_assignee) if prior_assignee else None ), "prior_borrowed_by": ( str(prior_borrower) if prior_borrower else None ), }, confirmed_at=datetime.now(UTC), confirmed_by=operator_id, ) self.db.add(operation) history = AssetHistory( asset_id=asset.id, action="recovered_from_leaver", from_user=prior_assignee or prior_borrower, operator_id=operator_id, notes=notes[:2000] if notes else "Recovered from ex-employee", ) self.db.add(history) asset.status = "idle" asset.assigned_to = None asset.assigned_at = None asset.borrowed_by = None asset.borrow_started_at = None asset.expected_return_date = None asset.updated_at = datetime.now(UTC) await self.db.flush() return asset, operation # ------------------------------------------------------------------ # Office-held mode (migration 183, 2026-05-19). # Companion to ``held_by_leaver`` for the "no Person assignee, lives # at a city office" case. ``recycle_to_office`` is the canonical way # to move a device into office_held; ``release_from_office`` is the # inverse (back to idle stock). The bulk variant + Excel intake + # offboarding T+0 disposition handler all funnel through # ``recycle_to_office`` so the invariant lives in exactly one place. # ------------------------------------------------------------------ async def _verify_office_active( self, office_id: UUID, ) -> AssetOfficeAddress: """Verify the office exists and is active. Raises ``ASSET_OFFICE_NOT_FOUND`` for unknown/archived offices. Used at every write entry point that sets ``holder_office_id`` so an invalid id never slips past the service into a NOT-NULL column (the FK is ON DELETE SET NULL — it cannot itself reject an insert of an inactive but live row). """ result = await self.db.execute( select(AssetOfficeAddress).where( AssetOfficeAddress.id == office_id, ) ) office = result.scalar_one_or_none() if office is None or not office.is_active: raise AppException( code="ASSET_OFFICE_NOT_FOUND", message=f"Office {office_id} not found or inactive", status_code=400, ) return office async def recycle_to_office( self, asset_id: UUID, *, office_id: UUID, reason: str, operator_id: UUID, source: str = "manual", ) -> tuple[AssetItem, AssetOperation]: """Recycle an asset to a city office. Allowed source states: * personal-held (idle / available / claimed / assigned) — transitions to ``held_by_office`` and clears the ``assigned_to`` reference (the prior holder is captured in the operation's ``from_user_id`` and audit history). * already ``held_by_office`` — idempotent no-op when the target office matches; otherwise we re-route to the new office (transferring between city offices is a legitimate operation, just doesn't need ceremony). Rejected source states: * ``borrowed`` — caller must ``return_borrowed_asset`` first; recycling under a borrower silently strands them with the device while the row says it lives at the office. * ``held_by_leaver`` — recover first (the leaver claim must be cleared explicitly so audit shows the device coming back from the ex-employee). * terminal states (``disposed`` / ``retired``) — never. ``source='manual'`` (the asset admin clicked recycle on the detail page) is the default. The offboarding T+0 advance handler calls with ``source='offboarding'`` so the operation detail can be filtered out of manual-recycle reports. """ result = await self.db.execute( select(AssetItem) .where(AssetItem.id == asset_id, AssetItem.deleted_at.is_(None)) .with_for_update() ) asset = result.scalar_one_or_none() if asset is None: raise NotFoundError("Asset", str(asset_id)) # Explicit whitelist of source states (PR-#406 review fix). A # prior draft used a blacklist (reject disposed/borrowed/leaver), # which silently accepted in-flight pending states # (claim_pending / return_pending / borrow_pending / # borrow_return_pending / transferring) and orphaned the in-flight # operations they correspond to. Mirror docstring exactly: # idle / available / claimed / assigned are valid sources; the # same-office re-recycle path is a no-op handled below. _ALLOWED_SOURCE_STATUSES = frozenset({ "idle", "available", "claimed", "assigned", ASSET_STATUS_HELD_BY_OFFICE, # may re-route between offices }) if asset.status in ("disposed", "disposal_pending"): raise AppException( code="ASSET_TERMINAL", message=( f"Asset is {asset.status}; cannot recycle to office." ), status_code=400, ) if asset.status == "borrowed" or asset.borrowed_by is not None: raise AppException( code="ASSET_BORROWED", message=( "Asset is currently on loan; return the borrow first." ), status_code=400, ) if asset.status == ASSET_STATUS_HELD_BY_LEAVER: raise AppException( code="ASSET_HELD_BY_LEAVER", message=( "Asset is held by an ex-employee; use recover_from_leaver " "first, then recycle to office." ), status_code=400, ) if asset.status not in _ALLOWED_SOURCE_STATUSES: # Pending / transferring / maintenance / faulty / testing / # legacy / etc — explicitly reject. Caller must resolve the # pending op or fix the physical state first. raise AppException( code="ASSET_NOT_RECYCLABLE", message=( f"Asset status {asset.status!r} is not a valid source " "for recycle_to_office. Resolve pending operations or " "physical state first." ), status_code=400, ) office = await self._verify_office_active(office_id) prior_status = asset.status prior_assignee = asset.assigned_to prior_office_id = asset.holder_office_id # Idempotent no-op when already routed to the same office. # PR-#406 review fix: do NOT insert a new AssetOperation row on # every repeat call — that turns a double-click into operation-log # spam. Return the most-recent matching operation if one exists, # otherwise synthesise an in-memory marker so the caller can still # render "已在 X 办公室" without persisting. if ( asset.status == ASSET_STATUS_HELD_BY_OFFICE and asset.holder_office_id == office_id ): existing = ( await self.db.execute( select(AssetOperation) .where( AssetOperation.asset_id == asset.id, AssetOperation.operation_type == "recycle_to_office", ) .order_by(AssetOperation.created_at.desc()) .limit(1) ) ).scalar_one_or_none() if existing is not None: return asset, existing # No prior recycle row (e.g. the row was migration-185 # backfilled into held_by_office and the backfill operation # was on a different asset_id key) — synthesise a transient # marker. We MUST persist this single time so the API # response has an operation_number to render; subsequent # repeat calls return THIS row instead of a fresh one. operation = AssetOperation( operation_number=_generate_operation_number(), asset_id=asset.id, operation_type="recycle_to_office", status="confirmed", from_user_id=prior_assignee, operator_id=operator_id, notes=f"[noop, already at office {office_id}] {reason}"[:2000], detail={ "source": source, "noop": True, "office_id": str(office_id), }, confirmed_at=datetime.now(UTC), confirmed_by=operator_id, ) self.db.add(operation) await self.db.flush() return asset, operation now = datetime.now(UTC) asset.status = ASSET_STATUS_HELD_BY_OFFICE asset.holder_kind = ASSET_HOLDER_KIND_OFFICE asset.holder_office_id = office_id # Sync free-text ``location`` to the office's city so inventory # group-bys, location filters, and the location_admin routing # fallback don't keep pointing at the prior holder's # location after the asset physically moved. The office's # ``city`` is operator-curated (asset_office_addresses) and # not FK-bound to ``asset_locations`` — route through the # resolver so a typo on the office side can't propagate. If # unresolvable, leave the existing value untouched (less # surprising than blanking the column on a sync step). _recycle_resolver = await self._get_location_resolver() _recycle_resolved = _recycle_resolver.resolve(office.city) if _recycle_resolved is not None: asset.location = _recycle_resolved asset.assigned_to = None asset.assigned_at = None # Office-held has no expected_return_date — clearing it stops # stale "due back" reports from flagging the office on a # previously claimed asset. asset.expected_return_date = None asset.updated_at = now operation = AssetOperation( operation_number=_generate_operation_number(), asset_id=asset.id, operation_type="recycle_to_office", status="confirmed", from_user_id=prior_assignee, operator_id=operator_id, notes=reason[:2000] if reason else None, detail={ "source": source, "office_id": str(office_id), "prior_status": prior_status, "prior_assigned_to": ( str(prior_assignee) if prior_assignee else None ), "prior_office_id": ( str(prior_office_id) if prior_office_id else None ), }, confirmed_at=now, confirmed_by=operator_id, ) self.db.add(operation) history = AssetHistory( asset_id=asset.id, action="recycle_to_office", from_user=prior_assignee, operator_id=operator_id, notes=reason[:2000] if reason else None, ) self.db.add(history) await self.db.flush() return asset, operation async def bulk_recycle_to_office( self, request: AssetBulkRecycleToOfficeRequest, *, operator_id: UUID, ) -> BulkAssetMutationResponse: """Bulk-recycle multiple assets to one office. Mirrors ``bulk_claim_assets`` / ``bulk_borrow_assets``: takes the Redis idempotency lock, then DB-side idempotency check, then per-row try/except so partial failure doesn't roll back the whole batch. Each successful row's ``AssetOperation.detail`` gets the idempotency key stamped on it so a repeat request is detected by the DB check above. """ if not await self._acquire_bulk_idempotency_lock( "recycle_to_office", request.idempotency_key, operator_id, ): raise AppException( code="DUPLICATE_BULK_OPERATION", message=( f"Bulk recycle with idempotency_key=" f"{request.idempotency_key} already in flight or completed" ), status_code=409, ) if await self._check_bulk_idempotency( "recycle_to_office", request.idempotency_key, operator_id, ): raise AppException( code="DUPLICATE_BULK_OPERATION", message=( f"Bulk recycle with idempotency_key=" f"{request.idempotency_key} already executed" ), status_code=409, ) await self._verify_office_active(request.office_id) results: list[BulkAssetRowResult] = [] successful = 0 for index, asset_id in enumerate(request.asset_ids, start=1): try: async with self.db.begin_nested(): asset, op = await self.recycle_to_office( asset_id, office_id=request.office_id, reason=request.reason, operator_id=operator_id, source="manual_bulk", ) # Reassign (not in-place mutate) so the plain JSON column # change is persisted — otherwise the DB idempotency guard # never sees the key. See the claim path for full rationale. if request.idempotency_key: op.detail = { **(op.detail or {}), "idempotency_key": request.idempotency_key, } successful += 1 results.append(BulkAssetRowResult( row_number=index, asset_tag=asset.asset_tag, asset_id=asset.id, status="held_by_office", operation_number=op.operation_number, )) except (AppException, ValueError) as exc: results.append(BulkAssetRowResult( row_number=index, asset_id=asset_id, status="failed", message=str(exc), )) return BulkAssetMutationResponse( total_rows=len(request.asset_ids), successful_rows=successful, failed_rows=len(request.asset_ids) - successful, results=results, ) async def release_from_office( self, asset_id: UUID, *, notes: str | None, operator_id: UUID, ) -> tuple[AssetItem, AssetOperation]: """Release an office-held asset back to idle stock. Inverse of ``recycle_to_office``. After release the asset is ``status='idle'`` / ``holder_kind='personal'`` with no holder references — ready to be ``claim_asset``-ed by a person. The release operation IS the audit trail that the office handed the device off (e.g. shipped to a remote new hire); no separate "ship-to" event is needed. """ result = await self.db.execute( select(AssetItem) .where(AssetItem.id == asset_id, AssetItem.deleted_at.is_(None)) .with_for_update() ) asset = result.scalar_one_or_none() if asset is None: raise NotFoundError("Asset", str(asset_id)) if asset.status != ASSET_STATUS_HELD_BY_OFFICE: raise AppException( code="ASSET_NOT_HELD_BY_OFFICE", message=( f"Asset is {asset.status}; release_from_office is only " "valid for assets currently in held_by_office." ), status_code=400, ) prior_office_id = asset.holder_office_id now = datetime.now(UTC) operation = AssetOperation( operation_number=_generate_operation_number(), asset_id=asset.id, operation_type="release_from_office", status="confirmed", from_user_id=None, operator_id=operator_id, notes=notes[:2000] if notes else "Released from office to idle", detail={ "prior_office_id": ( str(prior_office_id) if prior_office_id else None ), }, confirmed_at=now, confirmed_by=operator_id, ) self.db.add(operation) history = AssetHistory( asset_id=asset.id, action="release_from_office", from_user=None, operator_id=operator_id, notes=notes[:2000] if notes else "Released from office to idle", ) self.db.add(history) asset.status = "idle" asset.holder_kind = ASSET_HOLDER_KIND_PERSONAL asset.holder_office_id = None asset.assigned_to = None asset.assigned_at = None asset.updated_at = now await self.db.flush() return asset, operation async def resolve_office_for_person( self, person_id: UUID, ) -> AssetOfficeAddress | None: """Best-effort: which city office does this person work out of? Used by: * ``return_asset`` when ``return_target='office'`` is requested without an explicit office id (return to your home office). * Offboarding ``hardware_disposition`` default suggestion (the leaver's home office is the default recycle target). * The disposition preview endpoint. Resolution rules (in order — first hit wins): 1. Exact match on ``work_location_name`` → ``office.city``. 2. ``work_location_name`` with trailing ``市`` / ``省`` stripped (HRIS uses ``北京市``, asset offices use ``北京``). 3. Country-name fallback: when the person's ``work_location_name`` matches a country with exactly one office (Singapore, Japan, Malaysia, USA), use that office. Returns None when no plausible office can be inferred — the caller must decide what to do (typically: fall through to ``held_by_leaver`` or surface a "no home office" hint in the UI). """ from app.identity.person_models import IdentityPerson result = await self.db.execute( select(IdentityPerson.work_location_name) .where(IdentityPerson.id == person_id) ) loc_row = result.first() loc = loc_row[0] if loc_row else None if not loc: return None loc = loc.strip() if not loc: return None candidates = [loc] for suffix in ("市", "省", "县"): if loc.endswith(suffix): candidates.append(loc[:-1]) break offices = (await self.db.execute( select(AssetOfficeAddress).where( AssetOfficeAddress.is_active.is_(True), ) )).scalars().all() # Exact city match (and post-suffix-strip variant). for candidate in candidates: for office in offices: if office.city and office.city.strip() == candidate: return office # Country-name fallback (single-office country). by_country: dict[str, list[AssetOfficeAddress]] = {} for office in offices: if office.country_or_region: by_country.setdefault( office.country_or_region.strip(), [], ).append(office) for candidate in candidates: if candidate in by_country and len(by_country[candidate]) == 1: return by_country[candidate][0] return None # ------------------------------------------------------------------ # Operations management # ------------------------------------------------------------------ async def list_operations( self, page: int = 1, page_size: int = 20, asset_id: UUID | None = None, operation_type: str | None = None, status: str | None = None, keyword: str | None = None, operator_id: UUID | None = None, to_user_id: UUID | None = None, start_date: str | None = None, end_date: str | None = None, ) -> tuple[list[AssetOperation], int]: """List operations with filters.""" query = select(AssetOperation) count_query = select(func.count()).select_from(AssetOperation) conditions = [] if asset_id: conditions.append(AssetOperation.asset_id == asset_id) if operation_type: conditions.append(AssetOperation.operation_type == operation_type) if status: conditions.append(AssetOperation.status == status) if operator_id: conditions.append(AssetOperation.operator_id == operator_id) if to_user_id: conditions.append(AssetOperation.to_user_id == to_user_id) if keyword: # Join on asset to fuzzy-match asset_tag too. pattern = f"%{_esc(keyword)}%" query = query.join(AssetItem, AssetOperation.asset_id == AssetItem.id, isouter=True) count_query = count_query.join( AssetItem, AssetOperation.asset_id == AssetItem.id, isouter=True, ) conditions.append( or_( AssetOperation.operation_number.ilike(pattern), AssetItem.asset_tag.ilike(pattern), ) ) if start_date: try: start_dt = datetime.combine(date.fromisoformat(start_date), datetime.min.time()) except ValueError as exc: raise ValidationError( f"start_date must be ISO format (YYYY-MM-DD), got {start_date!r}", ) from exc conditions.append(AssetOperation.created_at >= start_dt) if end_date: try: end_dt = datetime.combine(date.fromisoformat(end_date), datetime.max.time()) except ValueError as exc: raise ValidationError( f"end_date must be ISO format (YYYY-MM-DD), got {end_date!r}", ) from exc conditions.append(AssetOperation.created_at <= end_dt) if conditions: query = query.where(and_(*conditions)) count_query = count_query.where(and_(*conditions)) total_result = await self.db.execute(count_query) total = total_result.scalar() or 0 offset = (page - 1) * page_size query = query.offset(offset).limit(page_size).order_by(AssetOperation.created_at.desc()) result = await self.db.execute(query) ops = list(result.scalars().all()) return ops, total async def confirm_operation( self, operation_id: UUID, action: str, operator_id: UUID, notes: str | None = None, ) -> AssetOperation: """Confirm or reject a pending operation.""" result = await self.db.execute( select(AssetOperation).where(AssetOperation.id == operation_id) ) operation = result.scalar_one_or_none() if not operation: raise NotFoundError("Operation", str(operation_id)) if operation.status != "pending": raise AppException( code="OPERATION_NOT_PENDING", message=f"Operation is {operation.status}, not pending", status_code=400, ) if action == "confirm": operation.status = "confirmed" else: operation.status = "rejected" operation.confirmed_at = datetime.now(UTC) operation.confirmed_by = operator_id if notes: operation.notes = (operation.notes or "") + f"\n[{action}] {notes}" await self.db.flush() return operation # ------------------------------------------------------------------ # Procurement orders # ------------------------------------------------------------------ async def create_procurement_order( self, data: ProcurementOrderCreateRequest, applicant_id: UUID, ) -> AssetProcurementOrder: """Create a procurement order.""" order = AssetProcurementOrder( order_number=_generate_order_number(), order_type=data.order_type, status="draft", applicant_id=applicant_id, supplier_id=data.supplier_id, department=data.department, supplier_name=data.supplier_name, total_amount=data.total_amount, items=data.items, notes=data.notes, requested_at=datetime.now(UTC), ) if data.supplier_id: supplier = await self.db.get(AssetSupplier, data.supplier_id) if not supplier: raise NotFoundError("Supplier", str(data.supplier_id)) order.supplier_name = supplier.name self.db.add(order) await self.db.flush() return order async def get_procurement_order(self, order_id: UUID) -> AssetProcurementOrder: """Get procurement order by ID.""" result = await self.db.execute( select(AssetProcurementOrder).where(AssetProcurementOrder.id == order_id) ) order = result.scalar_one_or_none() if not order: raise NotFoundError("ProcurementOrder", str(order_id)) return order async def update_procurement_order( self, order_id: UUID, data: ProcurementOrderUpdateRequest, ) -> AssetProcurementOrder: """Update a procurement order.""" order = await self.get_procurement_order(order_id) for field, value in data.model_dump(exclude_unset=True).items(): setattr(order, field, value) if data.supplier_id: supplier = await self.db.get(AssetSupplier, data.supplier_id) if not supplier: raise NotFoundError("Supplier", str(data.supplier_id)) order.supplier_name = supplier.name order.updated_at = datetime.now(UTC) await self.db.flush() return order async def list_procurement_orders( self, page: int = 1, page_size: int = 20, status: str | None = None, order_type: str | None = None, supplier_id: UUID | None = None, department: str | None = None, keyword: str | None = None, start_date: str | None = None, end_date: str | None = None, ) -> tuple[list[AssetProcurementOrder], int]: """List procurement orders.""" query = select(AssetProcurementOrder) count_query = select(func.count()).select_from(AssetProcurementOrder) conditions = [] if status: conditions.append(AssetProcurementOrder.status == status) if order_type: conditions.append(AssetProcurementOrder.order_type == order_type) if supplier_id: conditions.append(AssetProcurementOrder.supplier_id == supplier_id) if department: conditions.append(AssetProcurementOrder.department == department) if keyword: pattern = f"%{_esc(keyword)}%" conditions.append( or_( AssetProcurementOrder.order_number.ilike(pattern), AssetProcurementOrder.supplier_name.ilike(pattern), AssetProcurementOrder.notes.ilike(pattern), ) ) if start_date: try: start_dt = datetime.combine(date.fromisoformat(start_date), datetime.min.time()) except ValueError as exc: raise ValidationError( f"start_date must be ISO format (YYYY-MM-DD), got {start_date!r}", ) from exc conditions.append(AssetProcurementOrder.created_at >= start_dt) if end_date: try: end_dt = datetime.combine(date.fromisoformat(end_date), datetime.max.time()) except ValueError as exc: raise ValidationError( f"end_date must be ISO format (YYYY-MM-DD), got {end_date!r}", ) from exc conditions.append(AssetProcurementOrder.created_at <= end_dt) if conditions: query = query.where(and_(*conditions)) count_query = count_query.where(and_(*conditions)) total_result = await self.db.execute(count_query) total = total_result.scalar() or 0 offset = (page - 1) * page_size query = query.offset(offset).limit(page_size).order_by( AssetProcurementOrder.created_at.desc() ) result = await self.db.execute(query) orders = list(result.scalars().all()) return orders, total # ------------------------------------------------------------------ # Employee requests # ------------------------------------------------------------------ async def create_employee_request( self, data: EmployeeRequestCreateRequest, requester_id: UUID, ) -> AssetEmployeeRequest: """Create an employee request.""" req = AssetEmployeeRequest( request_number=_generate_request_number(), request_type=data.request_type, status="pending", requester_id=requester_id, asset_id=data.asset_id, category=data.category, target_user_id=data.target_user_id, reason=data.reason, detail=data.detail, ) self.db.add(req) await self.db.flush() return req async def get_employee_request( self, request_id: UUID, *, for_update: bool = False, ) -> AssetEmployeeRequest: """Get an employee request by ID. Pass ``for_update=True`` from the status-mutating paths (process / cancel) so the row is row-locked before the check-then-write, serializing concurrent approve-vs-cancel / double-click-approve (the rest of the asset domain locks the same way). Without it both writers read ``pending``, both pass the guard, and last-writer-wins → split-brain / duplicate audit rows. """ stmt = select(AssetEmployeeRequest).where( AssetEmployeeRequest.id == request_id ) if for_update: stmt = stmt.with_for_update() result = await self.db.execute(stmt) req = result.scalar_one_or_none() if not req: raise NotFoundError("EmployeeRequest", str(request_id)) return req async def process_employee_request( self, request_id: UUID, action: str, processor_id: UUID, notes: str | None = None, ) -> AssetEmployeeRequest: """Approve or reject an employee request.""" req = await self.get_employee_request(request_id, for_update=True) if req.status != "pending": raise AppException( code="REQUEST_NOT_PENDING", message=f"Request is {req.status}, not pending", status_code=400, ) # Pre-launch P1 fix (Round 45, CR-408): on approve, refuse # when the original requester has become inactive between # submit and admin decision. Without this an admin approves # an asset / borrow / buyback request whose requester is now # leaver_pending — the downstream claim_asset would fail but # the approval state is still committed pointing at a leaver # for forensic confusion. if action == "approve": requester = await self.db.get(AuthUser, req.requester_id) if requester is None or ( (requester.employment_status or "").lower() in {"left", "leaver_pending"} or requester.deleted_at is not None ): raise AppException( code="REQUESTER_INACTIVE", message=( "Cannot approve a request whose requester is " "no longer active — they may have begun " "offboarding since submitting." ), status_code=409, ) req.status = "approved" else: req.status = "rejected" req.processed_at = datetime.now(UTC) req.processed_by = processor_id if notes: req.notes = notes await self.db.flush() return req @staticmethod def _synthesize_snapshot_from_instance_form( feishu_detail: dict[str, Any], ) -> list[dict[str, Any]] | None: """Build a minimal form snapshot from the live Feishu instance. Fallback for legacy rows created before ``form_snapshot`` capture (2026-05-12): the widget list still lives in ``latest_instance_detail.form`` (a JSON string the detail endpoints refresh on open). Mirrors the frontend's ``synthesizeSnapshotFromFeishuForm`` — the panel renders from this same source, so refusing to validate against it produced the "panel shows upload widgets but submit 409s" asymmetry. Returns ``None`` when the blob has no usable form payload. """ instance = feishu_detail.get("latest_instance_detail") if not isinstance(instance, dict): return None form_raw = instance.get("form") if not isinstance(form_raw, str) or not form_raw: return None try: widgets = json.loads(form_raw) except (TypeError, ValueError): return None if not isinstance(widgets, list): return None entries: list[dict[str, Any]] = [] for widget in widgets: if not isinstance(widget, dict): continue wid = widget.get("id") if not isinstance(wid, str) or not wid: continue entries.append({ "id": wid, "name": str(widget.get("name") or ""), "type": str(widget.get("type") or "input"), "answer": None, }) return entries or None @staticmethod def _find_pending_self_task( instance_detail: dict[str, Any] | None, feishu_user_id: str, ) -> dict[str, Any] | None: """Sole PENDING ``task_list`` entry operated by the given Feishu user_id — the 提交人本人 node the laptop_replacement 回购 template parks on while waiting for the requester's upload (上传旧电脑恢复出厂设置 / 转账截图 nodes). Returns ``None`` when the flow is currently at someone else's node. Safety: if the requester is concurrently the operator on MORE than one pending node, return ``None`` rather than guessing. ``approve_task`` ADVANCES whichever node we pick, so picking the wrong one (e.g. a genuine approval-decision node where the requester also happens to be an approver) would silently approve a decision in their name — a four-eyes / self-approval bypass. The supplement material is never lost: an ambiguous case falls back to the comment channel. """ if not isinstance(instance_detail, dict): return None tasks = instance_detail.get("task_list") if not isinstance(tasks, list): return None self_tasks = [ task for task in tasks if isinstance(task, dict) and task.get("id") and str(task.get("status") or "").upper() == "PENDING" and str(task.get("user_id") or "") == feishu_user_id ] if len(self_tasks) != 1: return None return self_tasks[0] @staticmethod def _instance_form_value_counts( instance_detail: dict[str, Any] | None, ) -> dict[str, int]: """Per-widget value count parsed from the instance ``form`` blob (a JSON string of widget dicts). Upload widgets carry a list value — its length is the attached-file count. Used to verify a ``tasks/approve``+``form`` write actually landed: the count must grow by the number of supplemented files. Unparseable input yields ``{}`` (treated as "couldn't verify"). """ if not isinstance(instance_detail, dict): return {} form_raw = instance_detail.get("form") if not isinstance(form_raw, str) or not form_raw: return {} try: widgets = json.loads(form_raw) except (TypeError, ValueError): return {} if not isinstance(widgets, list): return {} counts: dict[str, int] = {} for widget in widgets: if not isinstance(widget, dict): continue wid = widget.get("id") if not isinstance(wid, str) or not wid: continue value = widget.get("value") if isinstance(value, list): counts[wid] = len(value) elif value in (None, "", {}): counts[wid] = 0 else: counts[wid] = 1 return counts # Comment-attachment ``type`` classifier. The v4 comment ``files`` # schema wants ``"image"`` for inline-rendered images and # ``"attachment"`` for everything else (videos included — Feishu # comments have no inline video tile). _IMAGE_FILENAME_RE = re.compile( r"\.(jpe?g|png|gif|bmp|webp)$", re.IGNORECASE, ) async def supplement_employee_request_uploads( self, *, request_id: UUID, actor: AuthUser, is_admin: bool, uploads: dict[str, list[SupplementUploadFile | str]], note: str | None, ) -> AssetEmployeeRequest: """Attach supplementary files to an in-flight approval. Built for the laptop_replacement 回购 path where the requester was allowed to submit without ``转账成功截图`` / ``上传恢复出厂 设置视频`` and attaches them later. Two delivery channels, best first: * **Form write** — the 回购 template parks the flow on 提交人本人 nodes whose upload widget has node-level "编辑" permission. When the actor is the requester and such a PENDING task exists, we approve it via ``tasks/approve`` with ``form`` carrying the upload codes: the files land in the form widget itself and the node completes (native-client parity). The write is verified by re-fetching the instance — per-widget value counts must grow by the supplemented file count. * **Comment fallback** — any widget the form write couldn't deliver (flow at someone else's node, admin acting on behalf, widget not editable at the current node, approve error) rides the instance-comment channel; approvers see the files in the approval-card timeline. Validation: * Caller must be the original requester (or any admin). * Request must be ``status='pending'`` — once approved / rejected / cancelled the Feishu instance is terminal and comments would not influence the decision. * The linked Feishu instance_code must exist in ``detail.feishu.instance_code``. * Every ``widget_id`` in ``uploads`` must appear in the form snapshot AND be of type ``image`` or ``attachmentV2``. Side effects: * Posts a comment with file/image attachments on the Feishu approval instance via :meth:`FeishuApprovalClient.add_instance_comment`. * Updates ``detail.feishu.form_snapshot`` so the My Requests detail drawer reflects the new tokens (audit ergonomics). * Appends a ``supplement_uploads`` audit trail entry under ``detail.feishu.supplements`` with timestamp + actor + widgets — so reviewers can see WHO supplemented WHAT WHEN without scrolling through audit_logs. """ req = await self.get_employee_request(request_id) if not is_admin and req.requester_id != actor.id: raise AppException( code="EMPLOYEE_REQUEST_NOT_OWNER", message="Only the original requester can supplement uploads.", status_code=403, ) if req.status != "pending": raise AppException( code="EMPLOYEE_REQUEST_INVALID_STATE", message=( f"Cannot supplement uploads on a request in " f"{req.status!r} state." ), status_code=409, ) detail = dict(req.detail or {}) feishu_detail = dict(detail.get("feishu") or {}) instance_code = feishu_detail.get("instance_code") if not instance_code: raise AppException( code="EMPLOYEE_REQUEST_NO_FEISHU_INSTANCE", message=( "This request is not linked to a Feishu approval " "instance; supplement upload is not supported." ), status_code=409, ) # Validate every widget against the captured form_snapshot. # The snapshot was recorded at submission time with the exact # set of widgets the applicant saw, so it's the authoritative # source for "does this widget id belong to this request, and # is it an upload widget". Legacy rows (pre-2026-05-12) have no # captured snapshot — synthesize one from # ``latest_instance_detail.form``, exactly like the frontend # does to render the panel; rejecting here while the panel # renders fine was the "no captured form snapshot" 409 the # admin hit on REQ-20260512-1A65. The synthesized snapshot is # persisted below (the mirror step), so the row self-heals # into the captured path on first supplement. snapshot_raw = feishu_detail.get("form_snapshot") if not isinstance(snapshot_raw, list) or not snapshot_raw: snapshot_raw = self._synthesize_snapshot_from_instance_form( feishu_detail, ) if not snapshot_raw: raise AppException( code="EMPLOYEE_REQUEST_NO_FORM_SNAPSHOT", message=( "This request has no captured form snapshot and no " "Feishu instance form to fall back on; supplement " "upload is not supported. 请先点「刷新流程」拉取飞书" "表单后重试。" ), status_code=409, ) snapshot_by_id: dict[str, dict[str, Any]] = {} for entry in snapshot_raw: if isinstance(entry, dict) and isinstance(entry.get("id"), str): snapshot_by_id[entry["id"]] = dict(entry) upload_widget_types = {"image", "attachmentV2"} # id → (widget name, widget type, [SupplementUploadFile, ...]). # Filename / url / object_key / size ride along (None for # legacy bare-token callers) so the Feishu comment can carry a # real, documented-schema attachment and the supplement history # stays viewable after the fact. validated: dict[ str, tuple[str, str, list[SupplementUploadFile]], ] = {} for widget_id, upload_entries in uploads.items(): entry = snapshot_by_id.get(widget_id) if entry is None: raise AppException( code="EMPLOYEE_REQUEST_UNKNOWN_WIDGET", message=( f"Widget {widget_id!r} is not part of this " "approval's form." ), status_code=400, ) wtype = str(entry.get("type") or "") if wtype not in upload_widget_types: raise AppException( code="EMPLOYEE_REQUEST_NOT_UPLOAD_WIDGET", message=( f"Widget {entry.get('name') or widget_id!r} is " f"of type {wtype!r}; only image / attachmentV2 " "widgets accept supplementary uploads." ), status_code=400, ) files: list[SupplementUploadFile] = [] for upload_entry in upload_entries: if isinstance(upload_entry, str): files.append(SupplementUploadFile(token=upload_entry)) else: files.append(upload_entry) validated[widget_id] = ( str(entry.get("name") or widget_id), wtype, files, ) # Idempotency: a replay of the same submit (double-click / client # retry / MCP re-invoke) carries the identical upload tokens # minted on the first call. Without this guard the form-write # node is gone (already approved) so every widget falls through # to the comment channel again, duplicating the approver's # timeline comment and double-appending tokens into the local # snapshot. If every incoming token already landed in a prior # supplement, the call is a no-op replay — return unchanged. prior_tokens: set[str] = set() for past in feishu_detail.get("supplements") or []: if not isinstance(past, dict): continue for up in (past.get("uploads") or {}).values(): if not isinstance(up, dict): continue for pf in up.get("files") or []: if isinstance(pf, dict) and isinstance(pf.get("token"), str): prior_tokens.add(pf["token"]) incoming_tokens = { f.token for _wid, (_n, _t, files) in validated.items() for f in files if f.token } if incoming_tokens and incoming_tokens <= prior_tokens: return req # Resolve the actor's Feishu user_id so the comment renders as # them (rather than as a generic bot post). On lookup failure # we surface a hard error — the comment endpoint REQUIRES a # user_id and silently swallowing it would leave the requester # thinking the upload synced when it did not. cfg = await get_feishu_config(session=self.db) client = FeishuApprovalClient( app_id=str(cfg.get("app_id") or ""), app_secret=str(cfg.get("app_secret") or ""), ) feishu_user_id = await client.resolve_user_id_by_email( actor.work_email, ) # ── Form-write path (preferred) ── # The laptop_replacement 回购 template parks the flow on # 提交人本人 nodes (上传旧电脑恢复出厂设置 / 转账截图) whose # upload widget carries node-level "编辑" permission. While the # flow sits on such a node the requester IS the pending task's # operator, so approving the task with ``form`` writes the # files into the widget itself and completes the node — exactly # what they'd do natively in the Feishu client. Only taken when # the actor is the requester (an admin acting on behalf must # not silently approve a node in the requester's name). # # The approve API's form behaviour is only documented for # condition-branch routing, so the write is VERIFIED by # re-fetching the instance and checking the widget's value # count grew by the supplemented file count. Widgets that # didn't stick (not editable at this node / fetch failed) fall # back to the comment channel below — material is never lost. form_written: set[str] = set() completed_task: dict[str, Any] | None = None if req.requester_id == actor.id: live_detail: dict[str, Any] | None = None try: live_detail = await client.get_instance_detail(instance_code) except Exception: logger.warning( "supplement form-write: instance detail fetch failed " "for %s; falling back to comment", instance_code, exc_info=True, ) self_task = self._find_pending_self_task( live_detail, feishu_user_id, ) approval_code = str( feishu_detail.get("approval_code") or (live_detail or {}).get("approval_code") or "", ) if self_task and approval_code: pre_counts = self._instance_form_value_counts(live_detail) # Merge prior codes into the value: approve+form may # REPLACE the widget value, and the answer can already # hold the submit-time / earlier-supplement upload # codes. http(s) strings are display URLs from # synthesized snapshots, never valid widget codes. form_entries: list[dict[str, Any]] = [] for wid, (_name, wtype, files) in validated.items(): merged: list[str] = [] existing = snapshot_by_id.get(wid, {}).get("answer") if isinstance(existing, str): existing = [existing] if isinstance(existing, list): merged.extend( tok for tok in existing if isinstance(tok, str) and tok and not tok.startswith(("http://", "https://")) ) merged.extend(f.token for f in files) form_entries.append( {"id": wid, "type": wtype, "value": merged}, ) approve_comment = "通过 Helix 补传材料 / Supplemented via Helix" if note: approve_comment += f"：{note}" try: await client.approve_task( approval_code=approval_code, instance_code=instance_code, user_id=feishu_user_id, task_id=str(self_task["id"]), form=json.dumps(form_entries, ensure_ascii=False), comment=approve_comment, ) except Exception: logger.warning( "supplement form-write: tasks/approve failed for " "instance %s task %s; falling back to comment", instance_code, self_task.get("id"), exc_info=True, ) else: completed_task = { "task_id": str(self_task.get("id")), "node_name": str(self_task.get("node_name") or ""), } post_detail: dict[str, Any] | None = None try: post_detail = await client.get_instance_detail( instance_code, ) except Exception: logger.warning( "supplement form-write: post-approve detail " "fetch failed for %s; treating widgets as " "unverified", instance_code, exc_info=True, ) post_counts = self._instance_form_value_counts( post_detail, ) for wid, (_name, _wtype, files) in validated.items(): grew_by = ( post_counts.get(wid, 0) - pre_counts.get(wid, 0) ) if grew_by >= len(files): form_written.add(wid) if isinstance(post_detail, dict): # Keep the stored progress view in sync — the # node just completed and the widget now holds # the files. feishu_detail["latest_instance_detail"] = post_detail # ── Comment fallback ── # Build one comment carrying every file that did NOT land in # the form. The v4 comment ``files`` schema is URL-based # (``{url, fileSize, title, type}``) — see # ``add_instance_comment``'s docstring for the prod evidence # that token-shaped attachments are silently dropped. Files # whose upload didn't return a viewable url (legacy bare-token # MCP callers) cannot be attached; they still appear in the # comment text summary and the supplement history, and we flag # the gap in the comment so the approver knows to ask. fallback = { wid: entry for wid, entry in validated.items() if wid not in form_written } if fallback: attachments: list[dict[str, Any]] = [] undeliverable = 0 widget_summaries: list[str] = [] for widget_id, (name, _wtype, files) in fallback.items(): widget_summaries.append(f"{name}（{len(files)}）") for idx, f in enumerate(files, start=1): if not f.url: undeliverable += 1 continue title = f.name or f"{name}-{idx}" entry_payload: dict[str, Any] = { "url": f.url, "title": title, "type": ( "image" if self._IMAGE_FILENAME_RE.search(title) else "attachment" ), } if f.size is not None: entry_payload["fileSize"] = f.size attachments.append(entry_payload) text_lines = [ "申请人补传了以下材料 / Requester supplemented:", *(f"- {summary}" for summary in widget_summaries), ] if undeliverable: text_lines.append( f"（{undeliverable} 个文件未能附带，请在 Helix 申请详情" "「补传材料」中查看 / file(s) not attached — view in Helix）" ) if note: text_lines.append("") text_lines.append(f"备注 / Note: {note}") comment_text = "\n".join(text_lines) await client.add_instance_comment( instance_code=instance_code, user_id=feishu_user_id, content_text=comment_text, attachments=attachments or None, ) # Mirror the supplement into our local form_snapshot so the # detail drawer can show "this widget now has files" without # an extra Feishu round-trip. Append to (rather than replace) # the answer so the multi-supplement case accumulates. new_snapshot = [] for entry in snapshot_raw: if not isinstance(entry, dict): new_snapshot.append(entry) continue entry_copy = dict(entry) wid = entry_copy.get("id") if isinstance(wid, str) and wid in validated: _name, _wtype, new_files = validated[wid] new_tokens = [f.token for f in new_files] existing = entry_copy.get("answer") if isinstance(existing, list): combined = [*existing, *new_tokens] elif existing in (None, "", [], {}): combined = list(new_tokens) else: combined = [existing, *new_tokens] entry_copy["answer"] = combined new_snapshot.append(entry_copy) # For legacy rows ``snapshot_raw`` is the synthesized fallback — # persisting it here backfills the row so later opens (and later # supplements) take the captured path directly. feishu_detail["form_snapshot"] = new_snapshot supplements = list(feishu_detail.get("supplements") or []) supplements.append({ "actor_id": str(actor.id), "actor_email": actor.work_email, "at": datetime.now(UTC).isoformat(), # How the files reached Feishu: "form" = every widget was # written into the instance form via the requester's own # pending-node approve; "comment" = all rode the comment # channel; "mixed" = some of each. ``completed_task`` names # the 提交人本人 node the form-write approved (null when # the comment path ran alone). "delivery": ( "form" if form_written and not fallback else "mixed" if form_written else "comment" ), "completed_task": completed_task, "uploads": { widget_id: { "name": name, "type": wtype, "count": len(files), "delivery": ( "form" if widget_id in form_written else "comment" ), # Per-file metadata — the ONLY place a supplemented # file stays reachable from Helix (comment # attachments never surface in # ``latest_instance_detail.form``). ``url`` is the # ephemeral signed Feishu link (~14h); # ``object_key`` is the durable Helix S3 copy the # presign endpoint serves long after that. "files": [ { "token": f.token, "name": f.name, "url": f.url, "object_key": f.object_key, "size": f.size, } for f in files ], } for widget_id, (name, wtype, files) in validated.items() }, "note": note, }) feishu_detail["supplements"] = supplements detail["feishu"] = feishu_detail req.detail = detail # SQLAlchemy doesn't auto-detect in-place JSON mutation; flag # the column as dirty so the UPDATE actually fires on flush. from sqlalchemy.orm.attributes import flag_modified flag_modified(req, "detail") await self.db.flush() return req async def cancel_employee_request( self, *, request_id: UUID, actor: AuthUser, is_admin: bool, ) -> AssetEmployeeRequest: """Cancel a still-pending employee request. Allowed for the original requester (or any admin) while ``status == 'pending'``. Tries to cancel the parallel Feishu approval instance too — a Feishu cancel failure is logged but does NOT block the local state change (admins fix orphans through Feishu's own UI rather than letting an approval API hiccup pin the row pending forever). Also marks any linked ``api_key_assignment`` row as cancelled so the assignment doesn't sit in "pending_approval" forever after the unified employee_request row was withdrawn. """ # Row-lock before the status check so a concurrent admin-approve # can't interleave with this cancel (approve-vs-cancel split-brain). req = await self.get_employee_request(request_id, for_update=True) if not is_admin and req.requester_id != actor.id: raise AppException( code="EMPLOYEE_REQUEST_NOT_OWNER", message="Only the original requester can cancel.", status_code=403, ) if req.status != "pending": raise AppException( code="EMPLOYEE_REQUEST_INVALID_STATE", message=f"Cannot cancel a request in {req.status!r} state.", status_code=409, ) # ``employee_requests`` doesn't carry an ``approval_request_id`` # column — the reverse linkage lives in # ``approval_requests.source_payload.employee_request_id``. # Look it up the same way the Feishu callback dispatcher does. feishu_instance_code: str | None = None approval_code: str = "" approval_request: ApprovalRequest | None = ( await self.db.execute( select(ApprovalRequest).where( func.json_unquote( func.json_extract( ApprovalRequest.source_payload, "$.employee_request_id", ) ) == str(req.id), ).order_by(ApprovalRequest.created_at.desc()).limit(1) ) ).scalar_one_or_none() if approval_request is not None: feishu_instance_code = approval_request.external_id approval_code = str( (approval_request.source_payload or {}).get("approval_code", "") ) self.last_feishu_cancel_synced = True if feishu_instance_code: try: cfg = await get_feishu_config(session=self.db) client = FeishuApprovalClient( app_id=str(cfg.get("app_id") or ""), app_secret=str(cfg.get("app_secret") or ""), ) feishu_user_id = await client.resolve_user_id_by_email( actor.work_email, ) await client.cancel_instance( instance_code=feishu_instance_code, user_id=feishu_user_id, approval_code=approval_code, ) except Exception as e: # noqa: BLE001 logger.warning( "Feishu cancel failed for employee_request %s " "(instance=%s): %s", req.id, feishu_instance_code, e, ) self.last_feishu_cancel_synced = False await notify_feishu_cancel_desync( self.db, flow_label="员工申请", resource_type="employee_request", resource_id=str(req.id), instance_code=feishu_instance_code, error=e, trace_id=f"employee_request_cancel:{req.id}", ) if approval_request is not None: # Race-safe terminal-state preservation. A Feishu # callback can land an "approved" / "rejected" outcome # while we were on the network call above — that # callback runs in its own session and committed # before we re-read here. Re-fetch under FOR UPDATE # so any concurrent callback's writes are visible AND # held under our row lock for the rest of the # transaction. Round-3 logic-scan P0 (2026-05-05). fresh_approval = await self.db.scalar( select(ApprovalRequest) .where(ApprovalRequest.id == approval_request.id) .with_for_update(), ) if fresh_approval is not None and fresh_approval.status == "pending": fresh_approval.status = "cancelled" fresh_approval.updated_at = datetime.now(UTC) approval_already_terminal = False else: logger.info( "cancel_employee_request: approval_request %s " "is in terminal state %r (callback raced); " "preserving the terminal state and refusing to " "flip the employee_request to cancelled", approval_request.id, fresh_approval.status if fresh_approval else "missing", ) approval_already_terminal = True else: approval_already_terminal = False else: approval_already_terminal = False # If the approval has reached a terminal state via a racing # callback, refuse to overwrite the employee_request — the # callback will have already activated/revoked the linked # asset and an admin clobber here would split state across # the two ledgers. Surface as a 409 so the user re-loads and # sees the actual outcome. Round-3 logic-scan P0. if approval_already_terminal: await self.db.refresh(req) raise AppException( code="EMPLOYEE_REQUEST_RACED_WITH_APPROVAL", message=( "Cannot cancel: the linked Feishu approval reached " f"a terminal state ({fresh_approval.status!r}) " "before your cancel landed. Refresh to see the " "actual outcome." ), status_code=409, ) # Re-read req under FOR UPDATE — the same callback that # terminalised approval_request also flipped req.status via # _apply_callback_state, so the in-memory copy is stale. fresh_req = await self.db.scalar( select(AssetEmployeeRequest) .where(AssetEmployeeRequest.id == req.id) .with_for_update(), ) if fresh_req is None or fresh_req.status != "pending": raise AppException( code="EMPLOYEE_REQUEST_INVALID_STATE", message=( f"Cannot cancel: employee_request reached " f"{fresh_req.status if fresh_req else 'missing'!r} " "while your cancel was in flight." ), status_code=409, ) req = fresh_req # Cascade: api_key_assignment rows linked to this employee # request must move out of "pending_approval" too — leaving a # dangling assignment row would be the kind of "looks fine in # the UI but the data is half-stale" bug we explicitly avoid. if req.request_type == "api_key_request": from app.domains.apikey.models import ApiKeyAssignment assignment_rows = ( await self.db.execute( select(ApiKeyAssignment).where( ApiKeyAssignment.employee_request_id == req.id, ApiKeyAssignment.status == "pending_approval", ) ) ).scalars().all() now_ts = datetime.now(UTC) for assignment in assignment_rows: assignment.status = "cancelled" assignment.updated_at = now_ts # Cascade: a pending SoftwareSeatAllocation linked to this # request must be revoked too. Without this, the row stays in # ``status='pending'`` after the user cancels, and # ``check_no_active_seat_request`` blocks them from re-applying # for the same plan ("You already have a pending seat for X"). # The Feishu callback path (``_revoke_pending_seat_on_rejection``) # already handles the case where Feishu fires a cancel webhook # back to us, but a user-initiated cancel that races ahead of # (or never gets) the Feishu callback orphaned the seat row. # Mirror the callback's revoke semantics: revoked_at = allocated_at # so the row's billable window collapses to zero. if req.request_type == "software_seat" and approval_request is not None: from app.domains.asset.models import SoftwareSeatAllocation seat_allocation_id = ( (approval_request.source_payload or {}).get("seat_allocation_id") ) if not seat_allocation_id: logger.warning( "software_seat cancel for request %s missing " "seat_allocation_id in approval source_payload — " "pending row (if any) will be left for the " "approval_pending_reaper cron to clean up", req.request_number, ) else: try: seat_uuid = UUID(str(seat_allocation_id)) except (ValueError, TypeError): # Malformed source_payload — only Helix code writes # this column so a bad value means a corrupted row. # Log + skip rather than raise: the cancel itself # still succeeds and the orphan can be cleaned by # the reaper cron. logger.warning( "software_seat cancel for request %s has " "malformed seat_allocation_id %r — skipping", req.request_number, seat_allocation_id, ) seat_uuid = None allocation = ( await self.db.scalar( select(SoftwareSeatAllocation).where( SoftwareSeatAllocation.id == seat_uuid, ), ) if seat_uuid is not None else None ) if allocation is not None and allocation.status == "pending": if allocation.user_id != req.requester_id: # Defence-in-depth — the approval_request must # belong to req.requester_id (validated upstream # by EMPLOYEE_REQUEST_NOT_OWNER), so a mismatched # seat user_id means source_payload was tampered # with at write time. Refuse and log for audit. logger.error( "software_seat cancel refused for %s: allocation " "user_id (%s) does not match requester (%s)", seat_allocation_id, allocation.user_id, req.requester_id, ) else: allocation.status = "revoked" allocation.revoked_at = allocation.allocated_at allocation.reason = ( (allocation.reason or "") + "\n[auto-revoked: requester cancelled the " "Feishu approval; revoked_at aligned to " "allocated_at so the row never enters a bill]" ).strip() req.status = "cancelled" req.processed_at = datetime.now(UTC) req.processed_by = actor.id await self.db.flush() return req async def list_employee_requests( self, page: int = 1, page_size: int = 20, requester_id: UUID | None = None, status: str | None = None, request_type: str | None = None, keyword: str | None = None, start_date: str | None = None, end_date: str | None = None, ) -> tuple[list[AssetEmployeeRequest], int]: """List employee requests.""" query = select(AssetEmployeeRequest) count_query = select(func.count()).select_from(AssetEmployeeRequest) conditions = [] if requester_id: conditions.append(AssetEmployeeRequest.requester_id == requester_id) if status: conditions.append(AssetEmployeeRequest.status == status) if request_type: conditions.append(AssetEmployeeRequest.request_type == request_type) if keyword: pattern = f"%{_esc(keyword)}%" conditions.append( or_( AssetEmployeeRequest.request_number.ilike(pattern), AssetEmployeeRequest.reason.ilike(pattern), ) ) if start_date: try: start_dt = datetime.combine(date.fromisoformat(start_date), datetime.min.time()) except ValueError as exc: raise ValidationError( f"start_date must be ISO format (YYYY-MM-DD), got {start_date!r}", ) from exc conditions.append(AssetEmployeeRequest.created_at >= start_dt) if end_date: try: end_dt = datetime.combine(date.fromisoformat(end_date), datetime.max.time()) except ValueError as exc: raise ValidationError( f"end_date must be ISO format (YYYY-MM-DD), got {end_date!r}", ) from exc conditions.append(AssetEmployeeRequest.created_at <= end_dt) if conditions: query = query.where(and_(*conditions)) count_query = count_query.where(and_(*conditions)) total_result = await self.db.execute(count_query) total = total_result.scalar() or 0 offset = (page - 1) * page_size query = query.offset(offset).limit(page_size).order_by( AssetEmployeeRequest.created_at.desc() ) result = await self.db.execute(query) requests = list(result.scalars().all()) return requests, total # ------------------------------------------------------------------ # Enrichment helpers (backward compatible) # ------------------------------------------------------------------ async def enrich_with_users( self, assets: list[AssetItem], *, viewer_scope_user_ids: set[UUID] | None = None, ) -> list[dict]: """Enrich assets with user information. Derives department (from identity_org_memberships) and user_entity (from identity_persons.company_name) via the assigned user. Round 6 audit fix-up 2026-05-04: optional ``viewer_scope_user_ids`` parameter. When set, ``assigned_to_name`` / ``_email`` and ``borrowed_by_name`` / ``_email`` are REDACTED for users outside the scope set — closes a PII leak where a manager viewing a subordinate's borrowed asset (or vice versa) could read the cross-department counterpart's display name + work email. Pass None (default) for admin / no-scope contexts where full enrichment is appropriate. """ if not assets: return [] asset_ids = [asset.id for asset in assets] refreshed_assets_result = await self.db.execute( select(AssetItem).where(AssetItem.id.in_(asset_ids)) ) refreshed_assets = {asset.id: asset for asset in refreshed_assets_result.scalars().all()} # Which office sits at each asset's location, for display only. # Derived through the managed-location resolver rather than the # ``office.asset_location`` bridge column this PR removes: the # office's city already answers where it is, and a second stored # copy of the relationship is a second thing to keep in sync. locations = { asset.location for asset in refreshed_assets.values() if asset.location } office_by_asset_location: dict[str, AssetOfficeAddress] = {} if locations: from app.domains.asset.location_resolver import ( # noqa: PLC0415 LocationIndex, ) index = await LocationIndex.load(self.db) offices = ( await self.db.execute( select(AssetOfficeAddress).where( AssetOfficeAddress.is_active.is_(True), ) ) ).scalars().all() office_by_location_id: dict[UUID, AssetOfficeAddress] = {} for office in offices: location_id = index.resolve_id(office.city) if location_id is not None: office_by_location_id.setdefault(location_id, office) for raw in locations: location_id = index.resolve_id(raw) resolved_office = ( office_by_location_id.get(location_id) if location_id is not None else None ) if resolved_office is not None: office_by_asset_location[raw] = resolved_office # Get all unique user IDs (assignees + borrowers). # Round 2b audit fix-up 2026-05-04: include ``borrowed_by`` so # the response can render the borrower's name for borrowed # assets — without this, post-115 borrowed rows show no holder. user_ids = set() ordered_assets = [refreshed_assets.get(asset.id, asset) for asset in assets] for asset in ordered_assets: if asset.assigned_to: user_ids.add(asset.assigned_to) if asset.borrowed_by: user_ids.add(asset.borrowed_by) # Fetch users with org_memberships loaded users_map = {} if user_ids: users_result = await self.db.execute( select(AuthUser) .options(selectinload(AuthUser.org_memberships)) .where(AuthUser.id.in_(user_ids)) ) for user in users_result.scalars(): users_map[user.id] = user # Enrich assets enriched = [] for asset in ordered_assets: asset_dict = { "id": asset.id, "asset_tag": asset.asset_tag, "asset_name": asset.asset_name, "category": asset.category, "brand": asset.brand, "model": asset.model, "serial_number": asset.serial_number, "status": asset.status, "device_status": asset.device_status, "source_type": asset.source_type, "order_number": asset.order_number, "purchase_date": asset.purchase_date, "purchase_cost": asset.purchase_cost, # Provenance must travel WITH the amount. ``AssetResponse`` # defaults this to ``recorded``, so omitting it here made # every estimated row read as a recorded invoice figure on # the detail surfaces — a silent default masking the real # value (caught in dev verification, 2026-09-05). "purchase_cost_source": asset.purchase_cost_source, "warranty_until": asset.warranty_until, "usage_limit_months": asset.usage_limit_months, "supplier_name": asset.supplier_name, "location": asset.location, "resolved_office_name": ( office_by_asset_location[asset.location].office_name if asset.location in office_by_asset_location else None ), "resolved_office_city": ( office_by_asset_location[asset.location].city if asset.location in office_by_asset_location else None ), "owner_company": asset.owner_company, "usage_company": asset.usage_company, "purchase_entity": asset.purchase_entity, "currency": asset.currency, "expected_return_date": asset.expected_return_date, "department": None, "user_entity": None, "admin_id": asset.admin_id, "assigned_to": asset.assigned_to, "assigned_at": asset.assigned_at, "borrowed_by": asset.borrowed_by, "borrow_started_at": asset.borrow_started_at, "disposal_type": asset.disposal_type, "disposed_at": asset.disposed_at, "disposal_photos": asset.disposal_photos, "image_urls": asset.image_urls, "custom_fields": asset.custom_fields, "epandian_id": asset.epandian_id, "notes": asset.notes, "last_inventory_at": asset.last_inventory_at, "last_inventory_cycle_id": asset.last_inventory_cycle_id, "last_inventory_status": asset.last_inventory_status, "created_at": asset.created_at, "updated_at": asset.updated_at, "deleted_at": asset.deleted_at, } if asset.assigned_to and asset.assigned_to in users_map: user = users_map[asset.assigned_to] if ( viewer_scope_user_ids is None or asset.assigned_to in viewer_scope_user_ids ): asset_dict["assigned_to_name"] = ( user.display_name or user.email.split("@")[0] ) asset_dict["assigned_to_email"] = user.email asset_dict["assigned_to_employee_no"] = user.employee_no asset_dict["assigned_to_employment_status"] = ( user.employment_status ) # Department + user_entity derive from the assigned # user; surface them only when name is shown. if ( viewer_scope_user_ids is None or asset.assigned_to in viewer_scope_user_ids ): asset_dict["department"] = user.department asset_dict["user_entity"] = user.company_name if asset.borrowed_by and asset.borrowed_by in users_map: borrower = users_map[asset.borrowed_by] if ( viewer_scope_user_ids is None or asset.borrowed_by in viewer_scope_user_ids ): asset_dict["borrowed_by_name"] = ( borrower.display_name or borrower.email.split("@")[0] ) asset_dict["borrowed_by_email"] = borrower.email asset_dict["borrowed_by_employee_no"] = borrower.employee_no asset_dict["borrowed_by_employment_status"] = ( borrower.employment_status ) enriched.append(asset_dict) return enriched async def enrich_history_with_users( self, history_items: list[AssetHistory], ) -> list[dict]: """Enrich history with user information.""" # Get all unique user IDs user_ids = set() for item in history_items: if item.from_user: user_ids.add(item.from_user) if item.to_user: user_ids.add(item.to_user) if item.operator_id: user_ids.add(item.operator_id) # Fetch users users_map = {} if user_ids: users_result = await self.db.execute( select(AuthUser).where(AuthUser.id.in_(user_ids)) ) for user in users_result.scalars(): users_map[user.id] = user # Enrich history enriched = [] for item in history_items: history_dict = { "id": item.id, "action": item.action, "from_user": item.from_user, "to_user": item.to_user, "operator_id": item.operator_id, "notes": item.notes, "created_at": item.created_at, } if item.operator_id and item.operator_id in users_map: user = users_map[item.operator_id] history_dict["operator_name"] = user.display_name or user.email.split("@")[0] enriched.append(history_dict) return enriched async def enrich_operations_with_users( self, operations: list[AssetOperation], *, admin_view: bool = True, ) -> list[dict]: """Enrich operations with operator/from/to user names. ``admin_view`` (default True for back-compat) gates the ``detail`` payload's HR-sourced fields. When False, the ``location_raw`` key (which carries the raw ``identity_persons.work_location_name`` an operator submitted before the resolver canonicalised it — e.g. ``Santa Clara``, ``北京市``, ``Würzburg``) is stripped from every operation's detail dict. Manager / employee / borrower views (i.e. ``GET /assets/{id}`` for any non-admin viewer that still passes the scope check via ``_row_user_in_scope``) MUST set ``admin_view=False`` so HR-internal location strings stay bounded to the admin surface. Round-3 review medium. """ operations = sorted( operations, key=lambda op: ( op.created_at or datetime.min.replace(tzinfo=UTC), op.operation_number, ), reverse=True, ) user_ids: set[UUID] = set() for op in operations: user_ids.add(op.operator_id) if op.from_user_id: user_ids.add(op.from_user_id) if op.to_user_id: user_ids.add(op.to_user_id) if op.confirmed_by: user_ids.add(op.confirmed_by) users_map: dict[UUID, AuthUser] = {} if user_ids: result = await self.db.execute( select(AuthUser).where(AuthUser.id.in_(user_ids)) ) for user in result.scalars(): users_map[user.id] = user def _name(uid: UUID | None) -> str | None: if uid and uid in users_map: u = users_map[uid] return u.display_name or u.email.split("@")[0] return None def _email(uid: UUID | None) -> str | None: if uid and uid in users_map: return users_map[uid].email return None def _emp_no(uid: UUID | None) -> str | None: if uid and uid in users_map: return users_map[uid].employee_no return None def _emp_status(uid: UUID | None) -> str | None: if uid and uid in users_map: return users_map[uid].employment_status return None def _redact_detail(detail: dict | None) -> dict | None: """Strip admin-only keys (``location_raw``) from detail when the viewer isn't an admin. Returns a NEW dict so the ORM row's detail JSON isn't mutated in place.""" if not detail or admin_view: return detail return {k: v for k, v in detail.items() if k != "location_raw"} enriched = [] for op in operations: op_dict = { "id": op.id, "operation_number": op.operation_number, "asset_id": op.asset_id, "operation_type": op.operation_type, "status": op.status, "from_user_id": op.from_user_id, "to_user_id": op.to_user_id, "operator_id": op.operator_id, "operator_name": _name(op.operator_id), "operator_email": _email(op.operator_id), "operator_employee_no": _emp_no(op.operator_id), "operator_employment_status": _emp_status(op.operator_id), "from_user_name": _name(op.from_user_id), "from_user_email": _email(op.from_user_id), "from_user_employee_no": _emp_no(op.from_user_id), "from_user_employment_status": _emp_status(op.from_user_id), "to_user_name": _name(op.to_user_id), "to_user_email": _email(op.to_user_id), "to_user_employee_no": _emp_no(op.to_user_id), "to_user_employment_status": _emp_status(op.to_user_id), "notes": op.notes, "detail": _redact_detail(op.detail), "created_at": op.created_at, "confirmed_at": op.confirmed_at, "confirmed_by": op.confirmed_by, } enriched.append(op_dict) return enriched # ----------------------------------------------------------------------- # Asset locations CRUD # ----------------------------------------------------------------------- async def list_locations(self, include_inactive: bool = False) -> list[AssetLocation]: stmt = select(AssetLocation) if not include_inactive: stmt = stmt.where(AssetLocation.is_active.is_(True)) stmt = stmt.order_by(AssetLocation.sort_order, AssetLocation.code) result = await self.db.execute(stmt) return list(result.scalars().all()) async def create_location(self, body: AssetLocationCreateRequest) -> AssetLocation: existing = await self.db.execute( select(AssetLocation).where(AssetLocation.code == body.code) ) if existing.scalar_one_or_none(): raise AppException( code="LOCATION_CODE_EXISTS", message=f"Location code '{body.code}' already exists", status_code=409, ) location = AssetLocation( code=body.code, name=body.name, sort_order=body.sort_order, ) self.db.add(location) await self.db.flush() return location async def update_location(self, location_id: UUID, body: AssetLocationUpdateRequest) -> AssetLocation: result = await self.db.execute( select(AssetLocation).where(AssetLocation.id == location_id) ) location = result.scalar_one_or_none() if not location: raise NotFoundError("Location not found") updates = body.model_dump(exclude_unset=True) if "code" in updates: existing = await self.db.execute( select(AssetLocation).where( AssetLocation.code == updates["code"], AssetLocation.id != location_id, ) ) if existing.scalar_one_or_none(): raise AppException( code="LOCATION_CODE_EXISTS", message=f"Location code '{updates['code']}' already exists", status_code=409, ) for key, value in updates.items(): setattr(location, key, value) await self.db.flush() return location async def delete_location(self, location_id: UUID) -> AssetLocation: result = await self.db.execute( select(AssetLocation).where(AssetLocation.id == location_id) ) location = result.scalar_one_or_none() if not location: raise NotFoundError("Location not found") # Round-27 pre-launch (2026-05-25): refuse hard-delete when any # asset still references this location. Otherwise the delete # strands every asset sitting there — the rows survive with a # ``location`` string no master row backs, which drops them out # of the location filter AND out of every regional # administrator's scope. Soft-disable via ``is_active=False`` is # the standard path; hard-delete is only for never-used rows. # # The predicate compared ``AssetItem.holder_office_id`` to the # location id until 2026-08-24. That column is an FK to # ``asset_office_addresses``, so it could never equal a location # id and the guard had never once fired. Assets reference a # location by name through ``AssetItem.location`` (deliberately # not an FK — see DECISIONS 2026-05-30), so that is what has to # be checked. in_use = (await self.db.execute( select(AssetItem.id).where( AssetItem.location == location.name, AssetItem.deleted_at.is_(None), ).limit(1), )).scalar_one_or_none() if in_use is not None: raise AppException( code="LOCATION_IN_USE", message=( "Cannot delete location: still referenced by active " "asset(s). Soft-disable via ``is_active=False`` to " "hide from new claims while preserving history." ), status_code=409, ) await self.db.delete(location) await self.db.flush() return location # ----------------------------------------------------------------------- # Asset category configs CRUD # ----------------------------------------------------------------------- async def list_categories(self, include_inactive: bool = False) -> list[AssetCategoryConfig]: stmt = select(AssetCategoryConfig) if not include_inactive: stmt = stmt.where(AssetCategoryConfig.is_active.is_(True)) stmt = stmt.order_by(AssetCategoryConfig.sort_order, AssetCategoryConfig.code) result = await self.db.execute(stmt) return list(result.scalars().all()) async def create_category(self, body: AssetCategoryConfigCreateRequest) -> AssetCategoryConfig: existing = await self.db.execute( select(AssetCategoryConfig).where(AssetCategoryConfig.code == body.code) ) if existing.scalar_one_or_none(): raise AppException( code="CATEGORY_CODE_EXISTS", message=f"Category code '{body.code}' already exists", status_code=409, ) category = AssetCategoryConfig( code=body.code, name=body.name, parent_code=body.parent_code, usage_limit_months=body.usage_limit_months, unit=body.unit, notes=body.notes, sort_order=body.sort_order, ) self.db.add(category) await self.db.flush() return category async def update_category(self, category_id: UUID, body: AssetCategoryConfigUpdateRequest) -> AssetCategoryConfig: result = await self.db.execute( select(AssetCategoryConfig).where(AssetCategoryConfig.id == category_id) ) category = result.scalar_one_or_none() if not category: raise NotFoundError("Category not found") updates = body.model_dump(exclude_unset=True) if "code" in updates: existing = await self.db.execute( select(AssetCategoryConfig).where( AssetCategoryConfig.code == updates["code"], AssetCategoryConfig.id != category_id, ) ) if existing.scalar_one_or_none(): raise AppException( code="CATEGORY_CODE_EXISTS", message=f"Category code '{updates['code']}' already exists", status_code=409, ) # A "system" category (``asset_category`` set — one of the canonical # rows mirroring ``ALLOWED_CATEGORIES``) backs the intake form's # fixed category dropdown AND its tag-code lookup. Deactivating it # would leave the option selectable but make every intake for that # type 400 with CATEGORY_TAG_CODE_MISSING — a silent footgun. Editing # its ``code`` is fine (the whole point); flipping it off is not. if category.asset_category and updates.get("is_active") is False: raise AppException( code="CATEGORY_SYSTEM_PROTECTED", message=( f"Cannot deactivate system category {category.code!r}: the " f"intake form's category list depends on it. Edit its code " f"instead." ), status_code=409, ) for key, value in updates.items(): setattr(category, key, value) await self.db.flush() return category async def delete_category(self, category_id: UUID) -> AssetCategoryConfig: result = await self.db.execute( select(AssetCategoryConfig).where(AssetCategoryConfig.id == category_id) ) category = result.scalar_one_or_none() if not category: raise NotFoundError("Category not found") # System categories (``asset_category`` set) back the intake form's # fixed dropdown + tag-code lookup — they must always exist, so block # deletion (admins edit the ``code``, never remove the row). Only # admin-created extra rows (no ``asset_category``) are deletable, and # those can't be referenced by any asset — ``asset_items.category`` # only ever holds a canonical enum value — so there's no orphan risk. if category.asset_category: raise AppException( code="CATEGORY_SYSTEM_PROTECTED", message=( f"Cannot delete system category {category.code!r}: the " f"intake form's category list depends on it." ), status_code=409, ) await self.db.delete(category) await self.db.flush() return category # ----------------------------------------------------------------------- # Asset tag generation (small sequence) # ----------------------------------------------------------------------- async def _category_tag_code(self, category: str) -> str: """Resolve a category enum (``laptop`` / ``desktop`` / ...) to its tag segment (``NB`` / ``DT`` / ...) from ``asset_category_configs``. The mapping is admin-managed via the 分类管理 tab — there is no hardcoded fallback. Fail loud if the active row is missing so a misconfigured category surfaces at intake rather than minting a garbage tag. """ cfg = (await self.db.execute( select(AssetCategoryConfig).where( AssetCategoryConfig.asset_category == category, AssetCategoryConfig.is_active.is_(True), ) )).scalar_one_or_none() if cfg is None or not cfg.code: raise AppException( code="CATEGORY_TAG_CODE_MISSING", message=f"No active category config for {category!r}; " f"set its 编码 in the 分类管理 admin tab", status_code=400, ) return cfg.code async def generate_asset_tag_for_category( self, purchase_entity_id: UUID, category: str, ) -> str: """Generate a tag from a category *enum* (resolves its code first). Thin wrapper over :meth:`generate_asset_tag` so both the intake create path and the ``/generate-tag`` endpoint share one category → code resolution. """ code = await self._category_tag_code(category) return await self.generate_asset_tag( purchase_entity_id=purchase_entity_id, category_code=code, ) async def generate_asset_tag( self, purchase_entity_id: UUID, category_code: str, ) -> str: """Generate next asset tag. Format: ``IT-{purchase_entity.code}-{category_code}-{YYYYMMDDHHMMSS}`` - ``purchase_entity.code`` comes from ``asset_legal_entities.code`` (managed via the OfficeAddress 公司主体信息 admin tab). Refuses (422) if the entity has no code so admins are forced to fill one in deliberately rather than auto-generating tags with ``None``-prefixed garbage. - ``category_code`` is the literal tag segment supplied by the caller (from ``asset_category_configs.code`` — e.g. ``DT`` for 台式机). Used verbatim. - Timestamp is the 14-digit ``YYYYMMDDHHMMSS`` wall clock in **Asia/Shanghai (UTC+8)** — the company's business timezone — NOT the server's UTC clock, so the printed asset number reads in Beijing time regardless of where the container runs. (Same convention as ``billing_service`` period anchoring.) To-the- second so two intakes for the same ``(entity, category)`` pair within the same minute don't collide; on the (rare) same-second collision we increment by 1s and retry up to 60 times, then raise rather than loop on a stuck clock. Example: ``IT-CN-NB-20260508151325``. """ entity = await self.db.get(AssetLegalEntity, purchase_entity_id) if entity is None: raise NotFoundError("Purchase entity", str(purchase_entity_id)) if not entity.is_active: raise AppException( code="LEGAL_ENTITY_INACTIVE", message="Purchase entity is inactive", status_code=400, ) if not entity.code: raise AppException( code="LEGAL_ENTITY_MISSING_CODE", message=( "Purchase entity has no code; set one in the " "公司主体信息 admin tab before recording assets" ), status_code=400, ) if not category_code: raise AppException( code="CATEGORY_CODE_INVALID", message="category_code must be non-empty", status_code=400, ) # Tag-collision retry budget. 60 ticks × 1s = a full minute; # past that we fail loudly instead of looping forever — the # only realistic way to exhaust 60 unique seconds is a stuck # clock or a same-second insert storm that needs a human. # Business timezone (UTC+8) so the embedded timestamp is Beijing # wall-clock, not the prod container's UTC. base_ts = datetime.now(ZoneInfo("Asia/Shanghai")) for offset in range(60): stamp = (base_ts + timedelta(seconds=offset)).strftime("%Y%m%d%H%M%S") candidate = f"IT-{entity.code}-{category_code}-{stamp}" existing = await self.db.execute( select(AssetItem.id).where(AssetItem.asset_tag == candidate) ) if existing.scalar_one_or_none() is None: return candidate raise AppException( code="ASSET_TAG_ALLOCATION_FAILED", message="Unable to allocate a unique asset tag after 60 attempts", status_code=400, )