"""Software one-time add-on charges — item config + request/approval + bill-line materialisation. An "add-on" is a one-time chargeable consumption on a software (AI credit packs, vendor person-days, one-off fees) that an employee buys through the same request→approval rails as a seat, but WITHOUT becoming a seat holder. Domain split: * Item CRUD + the request / cancel self-service flow live here. * The Feishu approval *callback* (leaver re-check + state flip) lives on ``ApprovalService`` next to the seat callback (``_approve_software_addon_charge``) to keep all approval side-effects in one place and avoid an import cycle. * The bill-line build for a period lives in ``billing_service`` next to the seat line build, so close / reopen / preview all pick add-ons up. Both the Feishu callback and the no-Feishu admin endpoint converge on the shared mutations ``mark_charge_approved`` / ``mark_charge_rejected`` here so the two approval paths can never drift. """ from __future__ import annotations import logging from datetime import UTC, datetime from decimal import ROUND_HALF_UP, Decimal from typing import Any from sqlalchemy import select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import AppException, NotFoundError from app.domains.asset.models import ( AssetSoftwareCatalog, SoftwareAddonCharge, SoftwareAddonItem, ) logger = logging.getLogger(__name__) _Q = Decimal("0.01") _REASON_MAX = 2000 def _quantize(value: Decimal) -> Decimal: return value.quantize(_Q, rounding=ROUND_HALF_UP) def validate_quantity(item: SoftwareAddonItem, quantity: int) -> None: """Enforce the item's quantity envelope. Raises AppException on breach. Mirrors the shape of ``check_plan_quota`` for seats: a single guard used by every entry point (HTTP, MCP) so the rules can't diverge. """ if not isinstance(quantity, int) or quantity <= 0: raise AppException( code="INVALID_QUANTITY", message="Quantity must be a positive integer.", status_code=400, ) if quantity < item.min_quantity: raise AppException( code="QUANTITY_BELOW_MIN", message=f"Minimum quantity for this item is {item.min_quantity}.", status_code=400, ) if item.max_quantity is not None and quantity > item.max_quantity: raise AppException( code="QUANTITY_ABOVE_MAX", message=f"Maximum quantity for this item is {item.max_quantity}.", status_code=400, ) step = item.step or 1 # Steps are measured from ``min_quantity`` so e.g. min=500, step=500 # admits 500/1000/1500 but rejects 750. if step > 1 and (quantity - item.min_quantity) % step != 0: raise AppException( code="QUANTITY_NOT_ON_STEP", message=( f"Quantity must be {item.min_quantity} plus a multiple of " f"{step}." ), status_code=400, ) def compute_amount(unit_price: Decimal, quantity: int) -> Decimal: return _quantize(Decimal(unit_price) * Decimal(quantity)) # ── Item catalog (admin) ──────────────────────────────────────────────── async def list_items_for_software( db: AsyncSession, software_id: Any, *, include_inactive: bool = True, ) -> list[SoftwareAddonItem]: stmt = select(SoftwareAddonItem).where( SoftwareAddonItem.software_id == software_id, ) if not include_inactive: stmt = stmt.where(SoftwareAddonItem.status == "active") stmt = stmt.order_by(SoftwareAddonItem.created_at.asc()) return list((await db.execute(stmt)).scalars().all()) async def list_active_items_all(db: AsyncSession) -> list[SoftwareAddonItem]: """Every ACTIVE add-on item across all software, ordered by software then item name. Powers the unified self-service request catalog where the employee hasn't pre-selected a software — they pick directly from the cross-software item list. ``software`` is lazy='joined' so the software name is loaded without an N+1. """ stmt = ( select(SoftwareAddonItem) .where(SoftwareAddonItem.status == "active") .order_by(SoftwareAddonItem.name.asc()) ) return list((await db.execute(stmt)).scalars().all()) async def create_item( db: AsyncSession, *, software_id: Any, name: str, unit_label: str, unit_price: Decimal, currency: str = "CNY", min_quantity: int = 1, step: int = 1, max_quantity: int | None = None, description: str | None = None, status: str = "active", ) -> SoftwareAddonItem: sw = (await db.execute( select(AssetSoftwareCatalog).where(AssetSoftwareCatalog.id == software_id) )).scalar_one_or_none() if not sw: raise NotFoundError("Software", str(software_id)) if unit_price is None or Decimal(unit_price) < 0: raise AppException( code="INVALID_UNIT_PRICE", message="Unit price must be a non-negative number.", status_code=400, ) if min_quantity < 1 or step < 1: raise AppException( code="INVALID_QUANTITY_ENVELOPE", message="min_quantity and step must be >= 1.", status_code=400, ) if max_quantity is not None and max_quantity < min_quantity: raise AppException( code="INVALID_QUANTITY_ENVELOPE", message="max_quantity must be >= min_quantity.", status_code=400, ) item = SoftwareAddonItem( software_id=software_id, name=name, unit_label=unit_label, unit_price=_quantize(Decimal(unit_price)), currency=(currency or "CNY").upper(), min_quantity=min_quantity, step=step, max_quantity=max_quantity, description=description, status=status if status in ("active", "inactive") else "active", ) db.add(item) await db.flush() # Populate the (lazy="joined") ``software`` relationship from the row we # already loaded for validation. Without this, the caller reading # ``item.software`` on this freshly-created object would trigger an # implicit lazy load and raise MissingGreenlet under the async session. item.software = sw return item async def update_item( db: AsyncSession, *, item_id: Any, payload: dict[str, Any], ) -> SoftwareAddonItem: item = (await db.execute( select(SoftwareAddonItem).where(SoftwareAddonItem.id == item_id) )).scalar_one_or_none() if not item: raise NotFoundError("Add-on item", str(item_id)) # ``unit_price`` / ``currency`` are frozen onto each charge at request # time, so editing them here only affects FUTURE charges — historical # and in-flight charges keep their snapshot. No usage-gate needed. if "unit_price" in payload and payload["unit_price"] is not None: if Decimal(payload["unit_price"]) < 0: raise AppException( code="INVALID_UNIT_PRICE", message="Unit price must be a non-negative number.", status_code=400, ) item.unit_price = _quantize(Decimal(payload["unit_price"])) for field in ("name", "unit_label", "description"): if field in payload and payload[field] is not None: setattr(item, field, payload[field]) if "currency" in payload and payload["currency"]: item.currency = payload["currency"].upper() if "min_quantity" in payload and payload["min_quantity"] is not None: if payload["min_quantity"] < 1: raise AppException( code="INVALID_QUANTITY_ENVELOPE", message="min_quantity must be >= 1.", status_code=400, ) item.min_quantity = payload["min_quantity"] if "step" in payload and payload["step"] is not None: if payload["step"] < 1: raise AppException( code="INVALID_QUANTITY_ENVELOPE", message="step must be >= 1.", status_code=400, ) item.step = payload["step"] if "max_quantity" in payload: item.max_quantity = payload["max_quantity"] if "status" in payload and payload["status"] in ("active", "inactive"): item.status = payload["status"] # Post-mutation envelope coherence check. if item.max_quantity is not None and item.max_quantity < item.min_quantity: raise AppException( code="INVALID_QUANTITY_ENVELOPE", message="max_quantity must be >= min_quantity.", status_code=400, ) await db.flush() return item # ── Shared approval-state mutations (Feishu callback + admin endpoint) ─── async def mark_charge_approved( db: AsyncSession, charge: SoftwareAddonCharge, *, department: str | None, ) -> None: """Idempotently flip a pending charge to approved and pin its billing period. Caller is responsible for the leaver / ownership re-checks and the audit row (the two callers diverge there).""" if charge.status != "pending": return # Pin to the current open billing period (Asia/Shanghai) — always a # not-yet-closeable period, so an approved charge can never miss its # period's close. See billing_service._current_period. from app.domains.asset.billing_service import _current_period charge.status = "approved" charge.approved_at = datetime.now(UTC) charge.period = _current_period() if department is not None: charge.department = department await db.flush() async def mark_charge_rejected( db: AsyncSession, charge: SoftwareAddonCharge, *, status: str = "rejected", ) -> None: """Idempotently move a pending charge to a terminal non-billable state. ``status`` is ``rejected`` or ``cancelled``; both collapse the charge so it is never materialised into a bill.""" if charge.status != "pending": return charge.status = status if status == "cancelled": charge.cancelled_at = datetime.now(UTC) else: charge.rejected_at = datetime.now(UTC) await db.flush() # ── Self-service request / cancel (HTTP + MCP) ────────────────────────── def _bound_reason(reason: str | None) -> str | None: if reason is None: return None reason = reason.strip() if len(reason) > _REASON_MAX: reason = reason[:_REASON_MAX] return reason or None async def request_charge( db: AsyncSession, *, item_id: Any, quantity: int, reason: str | None, current_user: Any, trace_id: str, idempotency_key: str, ip_address: str | None = None, user_agent: str | None = None, ) -> SoftwareAddonCharge: """Employee requests a one-time add-on charge. Mirrors ``asset.router.request_seat``: validate item + quantity, freeze the amount, create a ``pending`` charge, and submit to Feishu if a ``software_addon`` flow is configured (else fall back to admin review). Unlike seats there is NO per-(user, item) dedup — repeat purchases are legitimate. """ from app.approval.submission_idempotency import ( normalize_approval_idempotency_key, stable_approval_resource_id, ) from app.audit.service import AuditService reason = _bound_reason(reason) key = normalize_approval_idempotency_key(idempotency_key) charge_id = stable_approval_resource_id( scope="software_addon", applicant_id=current_user.id, idempotency_key=key, ) charge = await db.get(SoftwareAddonCharge, charge_id) item = (await db.execute( select(SoftwareAddonItem).where(SoftwareAddonItem.id == item_id) )).scalar_one_or_none() if not item: raise NotFoundError("Add-on item", str(item_id)) if item.status != "active": raise AppException( code="ITEM_INACTIVE", message="This add-on item is not available.", status_code=400, ) charge_created = charge is None if charge is not None: if ( charge.user_id != current_user.id or charge.item_id != item.id or charge.quantity != quantity or (charge.reason or None) != (reason or None) ): raise AppException( code="APPROVAL_IDEMPOTENCY_PAYLOAD_MISMATCH", message=( "This Idempotency-Key was already used with a different " "software add-on request." ), status_code=409, ) amount = charge.amount else: validate_quantity(item, quantity) amount = compute_amount(item.unit_price, quantity) candidate = SoftwareAddonCharge( id=charge_id, item_id=item.id, software_id=item.software_id, user_id=current_user.id, quantity=quantity, unit_price=item.unit_price, currency=item.currency, amount=amount, status="pending", reason=reason, department=getattr(current_user, "department", None), ) try: async with db.begin_nested(): db.add(candidate) await db.flush() charge = candidate except IntegrityError: charge = await db.get(SoftwareAddonCharge, charge_id) if charge is None: raise AppException( code="APPROVAL_SUBMISSION_IN_PROGRESS", message="This software add-on request is still in progress.", status_code=409, ) from None charge_created = False assert charge is not None # Populate the (lazy="joined") ``item`` relationship from the loaded row # so callers reading ``charge.item`` / ``charge.item.software`` on this # freshly-created charge don't trigger an implicit lazy load (which # raises MissingGreenlet under the async session). ``item`` was loaded # via a SELECT above, so its own ``software`` relationship is already # eager-loaded. charge.item = item sw = item.software software_name = sw.name if sw else "Unknown" from app.approval.service import ApprovalService approval_svc = ApprovalService(db) feishu_config = (await approval_svc.template_config()).get("software_addon") approval_request_id: str | None = None approval_link_created = False if feishu_config: from app.approval.schemas import AssetApprovalSubmitRequest lang = (getattr(current_user, "preferred_language", "") or "").lower() if lang.startswith("en") or lang.startswith("ja"): amount_str = f"{amount} {item.currency} ({quantity} × {item.unit_label})" else: amount_str = f"{amount} {item.currency}（{quantity} × {item.unit_label}）" # Bilingual answer keys — survive a Feishu widget rename in either # direction, exactly like the seat path. feishu_answers: dict[str, Any] = {} for value, aliases in ( (software_name, ("软件名称", "Software Name")), (item.name, ("增值项", "Add-on Item", "Item")), (str(quantity), ("数量", "Quantity")), (f"{item.unit_price} {item.currency}/{item.unit_label}", ("单价", "Unit Price", "Price")), (amount_str, ("金额", "Amount", "Total")), (reason or "", ("申请原因", "Reason", "Application Reason")), ): for alias in aliases: feishu_answers[alias] = value submit_body = AssetApprovalSubmitRequest( template_key="software_addon", reason=reason, answers=feishu_answers, ) approval_resp = await approval_svc.create_asset_approval_request( body=submit_body, current_user=current_user, trace_id=trace_id, ip_address=ip_address, user_agent=user_agent, idempotency_key=key, internal_idempotency_payload={ "kind": "software_addon", "item_id": str(item.id), "quantity": quantity, "reason": reason, }, internal_extra_source_data={"addon_charge_id": str(charge.id)}, internal_extra_detail={ "software_addon": { "software_id": str(item.software_id), "software_name": software_name, "item_id": str(item.id), "item_name": item.name, "unit_label": item.unit_label, "unit_price": str(item.unit_price), "quantity": quantity, "currency": item.currency, "amount": str(amount), "addon_charge_id": str(charge.id), }, }, ) approval_request_id = str(approval_resp.id) if charge.approval_request_id is None: charge.approval_request_id = approval_resp.id approval_link_created = True await db.flush() if charge_created or approval_link_created: await AuditService(db).log( trace_id=trace_id, actor_user_id=current_user.id, actor_type="user", action="software_addon.request", resource_type="software_addon_charge", resource_id=str(charge.id), status="success", detail={ "item_id": str(item.id), "item_name": item.name, "software_name": software_name, "quantity": quantity, "amount": str(amount), "currency": item.currency, "approval_request_id": approval_request_id, "feishu_approval": bool(feishu_config), }, ip_address=ip_address, ) return charge async def cancel_charge( db: AsyncSession, *, charge_id: Any, current_user: Any, trace_id: str, ip_address: str | None = None, ) -> SoftwareAddonCharge: """Owner cancels their own still-pending charge. Approved charges are spent money and cannot be self-cancelled.""" from app.audit.service import AuditService charge = (await db.execute( select(SoftwareAddonCharge).where(SoftwareAddonCharge.id == charge_id) )).scalar_one_or_none() if not charge: raise NotFoundError("Add-on charge", str(charge_id)) if charge.user_id != current_user.id: raise AppException( code="FORBIDDEN", message="You can only cancel your own add-on requests.", status_code=403, ) if charge.status != "pending": raise AppException( code="INVALID_STATUS", message=( f"Cannot cancel a charge in {charge.status!r} status — only " "pending requests are cancellable." ), status_code=400, ) await mark_charge_rejected(db, charge, status="cancelled") # We intentionally do NOT chase the parallel Feishu card here: if it is # later approved, ``_approve_software_addon_charge`` finds the charge in # ``cancelled`` status and ``mark_charge_approved`` no-ops (the pending # guard), so a late approval can never resurrect or bill a cancelled # charge. The worst case is a stale card in the approver's inbox — a # cosmetic wart, not a money/consistency bug. await AuditService(db).log( trace_id=trace_id, actor_user_id=current_user.id, actor_type="user", action="software_addon.cancel", resource_type="software_addon_charge", resource_id=str(charge.id), status="success", detail={"item_id": str(charge.item_id)}, ip_address=ip_address, ) return charge