"""Finance budget lifecycle — plan → actual → variance → forecast → alert. The closed loop on top of the unified spend ledger. ``finance_budgets`` holds annual USD ceilings scoped by an optional ``(cost_type, department, cost_center)`` tuple; this service: * CRUD over the budgets (admin), * ``budget_vs_actual`` — measures each budget against the materialised ``finance_spend_facts`` actuals for a fiscal year as of a period, with YTD variance, burn %, and a run-rate year-end forecast, * ``evaluate_and_alert`` — folded into the daily ``finance.spend_rollup`` cron; fires a single de-duped burn-rate alert to the budget OWNER (a cost-centre owner / manager — never a mass employee push) when YTD burn crosses 80 % / 100 % of the YTD-prorated budget. Actuals match a budget by exact equality on whichever dimensions the budget sets (NULL = "all"). Budgets are enforced non-overlapping under a per-fiscal-year advisory lock. Historical conflicts remain visible for manual remediation but are excluded from aggregate totals and alerts. """ from __future__ import annotations import logging from collections.abc import AsyncIterator, Awaitable, Callable from contextlib import asynccontextmanager from decimal import ROUND_HALF_UP, Decimal, DecimalException from sqlalchemy import or_, select, text from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine, AsyncSession from app.core import fiscal_year as fy from app.core.exceptions import AppException from app.core.txn import best_effort from app.domains.finance.models import ( BASIS_CONSUMPTION, COST_TYPE_CLOUD, COST_TYPES, FinanceBudget, FinanceSpendFact, ) logger = logging.getLogger(__name__) _USD = Decimal("0.01") _ALERT_THRESHOLDS = (100, 80) # checked high→low; first crossed wins _NON_CLOUD_COST_TYPES = frozenset(COST_TYPES) - {COST_TYPE_CLOUD} _BUDGET_LOCK_TIMEOUT_SECONDS = 5 _MAX_BUDGET_AMOUNT = Decimal("99999999999999.99") def _q2(x) -> Decimal: return Decimal(str(x or 0)).quantize(_USD, rounding=ROUND_HALF_UP) def _validated_budget_amount(value: Decimal | str | None) -> Decimal: """Return a finite value representable by ``DECIMAL(16, 2)``.""" try: amount = Decimal(str(value)) except Exception as exc: raise AppException( "INVALID_AMOUNT", "amount_usd must be a finite decimal", status_code=422, ) from exc if not amount.is_finite() or amount < 0: raise AppException( "INVALID_AMOUNT", "amount_usd must be a finite value >= 0", status_code=422, ) try: amount = amount.quantize(_USD, rounding=ROUND_HALF_UP) except DecimalException as exc: raise AppException( "INVALID_AMOUNT", "amount_usd must be a finite decimal", status_code=422, ) from exc if amount > _MAX_BUDGET_AMOUNT: raise AppException( "INVALID_AMOUNT", f"amount_usd must not exceed {_MAX_BUDGET_AMOUNT}", status_code=422, ) return amount def _withheld_meta(withhold_cloud: bool) -> list[str]: """Declaration rider for a cloud-withheld budget payload (#2216). Imported lazily: ``read_service`` imports this module for the budget block on the leadership dashboard, so a module-level import here would close the cycle. """ from app.domains.finance.read_service import withheld_meta return withheld_meta(withhold_cloud) def _scope_filters(b: FinanceBudget): """SQL conditions matching facts in a budget's dimensional scope.""" conds = [] if b.cost_type: conds.append(FinanceSpendFact.cost_type == b.cost_type) if b.department: conds.append(FinanceSpendFact.department == b.department) if b.cost_center: conds.append(FinanceSpendFact.cost_center == b.cost_center) return conds def _scope_label(b: FinanceBudget) -> str: bits = [ b.cost_type or "全部类型", b.department or "全部门", b.cost_center or "全部成本中心", ] return " · ".join(bits) def _normalise_scope_value(value: str | None) -> str | None: """Collapse blank form values to the wildcard NULL representation.""" return value.strip() or None if value is not None else None def _scopes_overlap( left: FinanceBudget, right: FinanceBudget, ) -> bool: """True when two wildcard tuples can match at least one same fact.""" return all( left_value is None or right_value is None or left_value == right_value for left_value, right_value in ( (left.cost_type, right.cost_type), (left.department, right.department), (left.cost_center, right.cost_center), ) ) def diagnose_scope_conflicts( budgets: list[FinanceBudget], ) -> tuple[list[dict], set[int]]: """Diagnose legacy overlap without deleting or choosing a winner.""" conflicts: list[dict] = [] conflicted_ids: set[int] = set() for index, left in enumerate(budgets): for right in budgets[index + 1:]: if not _scopes_overlap(left, right): continue pair = sorted((left.id, right.id)) conflicts.append({ "budget_ids": pair, "reason": "wildcard_scope_overlap", }) conflicted_ids.update(pair) conflicts.sort(key=lambda item: item["budget_ids"]) return conflicts, conflicted_ids @asynccontextmanager async def _budget_lock_connection(db: AsyncSession) -> AsyncIterator[AsyncConnection]: """Keep one physical connection checked out for the named-lock lifetime.""" bind = db.bind if isinstance(bind, AsyncEngine): async with bind.connect() as connection: yield connection return if isinstance(bind, AsyncConnection): # A caller-bound connection is already held outside Session commit. yield bind return raise AppException( "BUDGET_LOCK_INFRASTRUCTURE_ERROR", "Budget scope lock infrastructure is unavailable; retry later", status_code=503, ) async def _acquire_budget_lock(db: AsyncConnection, fiscal_year: int) -> str: """Serialize the overlap check + commit for one fiscal year.""" lock_name = f"helix.finance.budget.{fiscal_year}" try: acquired = ( await db.execute( text("SELECT GET_LOCK(:name, :timeout)"), {"name": lock_name, "timeout": _BUDGET_LOCK_TIMEOUT_SECONDS}, ) ).scalar() except Exception as exc: logger.exception("Finance budget lock acquisition failed for FY%s", fiscal_year) raise AppException( "BUDGET_LOCK_INFRASTRUCTURE_ERROR", "Budget scope lock infrastructure is unavailable; retry later", status_code=503, ) from exc if acquired is None: raise AppException( "BUDGET_LOCK_INFRASTRUCTURE_ERROR", "Budget scope lock infrastructure is unavailable; retry later", status_code=503, ) if acquired != 1: raise AppException( "BUDGET_UPDATE_IN_PROGRESS", "Another budget update is in progress for this fiscal year", status_code=409, ) return lock_name async def _release_budget_lock(db: AsyncConnection, lock_name: str) -> None: # Called only from `finally:` blocks, right after `except: rollback; raise`. # A SAVEPOINT would buy nothing (a named lock is CONNECTION-scoped and # unaffected by one) and `best_effort` re-raises 1205/1213/1305 — from a # `finally`, that REPLACES the propagating original exception. Exactly the # call already made for `billing_service.close_month`'s RELEASE_LOCK. # savepoint-exempt: savepoint would mask the propagating exception try: released = (await db.execute( text("SELECT RELEASE_LOCK(:name)"), {"name": lock_name}, )).scalar() if released != 1: logger.error( "Failed to release finance budget lock %s on its owner connection " "(result=%r)", lock_name, released, ) except Exception: # noqa: BLE001 # MySQL releases named locks when the physical connection closes. # Never mask the mutation's real success/failure with cleanup noise. logger.error( "Failed to release finance budget lock %s; connection close is fallback", lock_name, ) # ─── CRUD ─────────────────────────────────────────────────────── async def _assert_scope_free(db: AsyncSession, *, fiscal_year: int, cost_type: str | None, department: str | None, cost_center: str | None, exclude_id: int | None = None) -> None: """Reject any wildcard scope overlapping another budget in the FY. For each dimension, NULL means every value, so an existing row conflicts when it is NULL or equals the proposed concrete value. If the proposed value is NULL the dimension imposes no filter and therefore overlaps all existing values. Production callers hold the per-FY advisory lock until commit, making this check safe under concurrent nullable inserts. """ stmt = select(FinanceBudget.id).where(FinanceBudget.fiscal_year == fiscal_year) for column, value in ( (FinanceBudget.cost_type, cost_type), (FinanceBudget.department, department), (FinanceBudget.cost_center, cost_center), ): if value is not None: stmt = stmt.where(or_(column.is_(None), column == value)) if exclude_id is not None: stmt = stmt.where(FinanceBudget.id != exclude_id) if (await db.execute(stmt.limit(1))).scalar_one_or_none() is not None: raise AppException( "BUDGET_SCOPE_CONFLICT", "该预算范围与同财年的现有预算重叠（NULL 表示全范围）", status_code=409, ) async def list_budgets(db: AsyncSession, fiscal_year: int) -> list[FinanceBudget]: return list((await db.execute( select(FinanceBudget).where(FinanceBudget.fiscal_year == fiscal_year) .order_by(FinanceBudget.department, FinanceBudget.cost_center) )).scalars().all()) async def create_budget(db: AsyncSession, *, fiscal_year: int, amount_usd: Decimal, cost_type: str | None = None, department: str | None = None, cost_center: str | None = None, owner_person_id: str | None = None, note: str | None = None) -> FinanceBudget: amount_usd = _validated_budget_amount(amount_usd) cost_type = _normalise_scope_value(cost_type) department = _normalise_scope_value(department) cost_center = _normalise_scope_value(cost_center) # Normalise blanks to NULL so wildcard semantics are stable. await _assert_scope_free( db, fiscal_year=fiscal_year, cost_type=cost_type, department=department, cost_center=cost_center, ) row = FinanceBudget( fiscal_year=fiscal_year, amount_usd=amount_usd, cost_type=cost_type, department=department, cost_center=cost_center, owner_person_id=(owner_person_id or None), note=(note or None), ) db.add(row) await db.flush() return row async def update_budget(db: AsyncSession, budget_id: int, **fields) -> FinanceBudget: row = await db.get(FinanceBudget, budget_id) if row is None: raise AppException("BUDGET_NOT_FOUND", "budget not found", status_code=404) if "amount_usd" in fields and fields["amount_usd"] is not None: amt = _validated_budget_amount(fields["amount_usd"]) # Raising the ceiling re-arms the alert ladder (burn % drops). if amt > row.amount_usd: row.last_alerted_threshold = 0 row.amount_usd = amt # If any scope dimension is being re-set, re-check the unique tuple. if any(k in fields for k in ("cost_type", "department", "cost_center")): proposed_scope = { key: _normalise_scope_value(fields.get(key, getattr(row, key))) for key in ("cost_type", "department", "cost_center") } await _assert_scope_free( db, fiscal_year=row.fiscal_year, cost_type=proposed_scope["cost_type"], department=proposed_scope["department"], cost_center=proposed_scope["cost_center"], exclude_id=row.id, ) fields.update(proposed_scope) # The marker describes both the measured scope and its recipient. A new # owner has not received the prior owner's alert, and a changed scope is # a different pool even when it happens to cross the same threshold. # Re-arm only on an actual identity/scope change; note-only and no-op # patches retain the existing de-dup state. for k in ("cost_type", "department", "cost_center", "owner_person_id"): if k not in fields: continue new_value = fields[k] or None if k == "owner_person_id" and new_value is not None: new_value = str(new_value) if getattr(row, k) != new_value: row.last_alerted_threshold = 0 for k in ("cost_type", "department", "cost_center", "owner_person_id", "note"): if k in fields: value = fields[k] or None if k == "owner_person_id" and value is not None: value = str(value) setattr(row, k, value) await db.flush() return row async def create_budget_atomic( db: AsyncSession, *, before_commit: Callable[[FinanceBudget], Awaitable[None]] | None = None, **fields, ) -> FinanceBudget: """Serialize overlap check, audit callback, and commit as one unit.""" fiscal_year = int(fields["fiscal_year"]) # Authentication and RBAC resolve on this same request session, opening # a REPEATABLE READ transaction before the route reaches us. End that # read snapshot before waiting for the FY lock; otherwise a later overlap # query can miss the preceding lock owner's committed insert. Commit (not # rollback) preserves the eagerly-loaded current-user object and persists # legitimate PAT last-used bookkeeping performed during authentication. await db.commit() async with _budget_lock_connection(db) as lock_connection: lock_name = await _acquire_budget_lock(lock_connection, fiscal_year) try: row = await create_budget(db, **fields) if before_commit is not None: await before_commit(row) await db.commit() return row except Exception: await db.rollback() raise finally: await _release_budget_lock(lock_connection, lock_name) async def update_budget_atomic( db: AsyncSession, budget_id: int, *, before_commit: Callable[[FinanceBudget], Awaitable[None]] | None = None, **fields, ) -> FinanceBudget: """Lock its FY through the combined mutation/audit commit.""" fiscal_year = await db.scalar( select(FinanceBudget.fiscal_year).where(FinanceBudget.id == budget_id) ) if fiscal_year is None: raise AppException("BUDGET_NOT_FOUND", "budget not found", status_code=404) # Release the pre-lock lookup/authentication snapshot and its pooled # business connection. The next budget read begins only after the named # lock is owned, so it observes the previous updater's commit. await db.commit() async with _budget_lock_connection(db) as lock_connection: lock_name = await _acquire_budget_lock(lock_connection, fiscal_year) try: updated = await update_budget(db, budget_id, **fields) if before_commit is not None: await before_commit(updated) await db.commit() return updated except Exception: await db.rollback() raise finally: await _release_budget_lock(lock_connection, lock_name) async def delete_budget(db: AsyncSession, budget_id: int) -> None: row = await db.get(FinanceBudget, budget_id) if row is None: raise AppException("BUDGET_NOT_FOUND", "budget not found", status_code=404) await db.delete(row) await db.flush() # ─── Budget vs actual (the read) ──────────────────────────────── async def _actuals_for_year( db: AsyncSession, fiscal_year: int, start_month: int, *, withhold_cloud: bool = False, ) -> list[FinanceSpendFact]: """One scan of the fiscal year's facts, reused across all budgets. The window is the fiscal year's Apr→Mar (per ``start_month``) span, not the calendar Jan→Dec — see :mod:`app.core.fiscal_year`. Lexical ``period`` comparison is correct across the calendar-year boundary the fiscal year straddles. Restricted to the CONSUMPTION basis. A finance budget is an annual *consumption* plan (cloud / software / ai / expense) — distinct from ``procurement_budgets`` which gates cash/PO commitments on the procurement basis. The ledger's hard invariant (``finance.models``) is that the two bases describe the same money twice and must never be summed; without this filter a budget scoped to ``cost_type=NULL`` (all types) for a department pulled in BOTH consumption facts AND the department's procurement-basis PO facts, inflating ``ytd_actual`` and firing false over-budget alerts at the owner. """ lo, hi = fy.fy_bounds(fiscal_year, start_month) stmt = select( FinanceSpendFact.period, FinanceSpendFact.cost_type, FinanceSpendFact.department, FinanceSpendFact.cost_center, FinanceSpendFact.amount_usd, ).where( FinanceSpendFact.period >= lo, FinanceSpendFact.period <= hi, FinanceSpendFact.basis == BASIS_CONSUMPTION, ) # #2216: budgets are measured against the SAME ledger the cloud # confidentiality gate covers. An out-of-circle reader who sees a # cloud-scoped budget's ``ytd_actual`` has the invoice figure directly; # on the company totals it is a subtraction away (total minus the three # axes they can still see). Withheld rows simply do not count towards # any budget's actual for that reader, and the payload declares it. from app.domains.finance.read_service import _withhold stmt = _withhold(stmt, withhold_cloud) return list((await db.execute(stmt)).all()) def _fact_matches(fact, b: FinanceBudget) -> bool: if b.cost_type and fact.cost_type != b.cost_type: return False if b.department and fact.department != b.department: return False if b.cost_center and fact.cost_center != b.cost_center: return False return True def _commit_matches(c: dict, b: FinanceBudget) -> bool: """A PO commitment matches a budget on department / cost_center (POs carry no clean cost_type, so a cost_type-scoped budget matches every open PO in its dept/cost-centre — documented limitation; most budgets are dept / cost-centre scoped). A budget with NO dept/cost_center scope (company-level or cost_type-only) has no PO axis to match on — matching every open PO would inflate its committed / forecast / commit_burn_pct and mis-fire ``status='over'``. Such budgets carry no commitments here.""" if not b.department and not b.cost_center: return False if b.department and c.get("department") != b.department: return False if b.cost_center and c.get("cost_center") != b.cost_center: return False return True async def budget_vs_actual(db: AsyncSession, fiscal_year: int, as_of_period: str | None = None, *, owner_person_id: str | None = None, start_month: int | None = None, withhold_cloud: bool = False) -> dict: """Measure every budget for ``fiscal_year`` against actuals as of a period. The fiscal year maps to its configured Apr→Mar window (see :mod:`app.core.fiscal_year`); ``as_of_period`` defaults to the current month clamped to the FY's last month so the run-rate forecast is meaningful mid-year. ``owner_person_id`` restricts to budgets that person owns — the self-scoped accountability view (a budget owner is an arbitrary person FK who may hold no finance role, so the owner-scoped read carries no admin gate at the router). """ if start_month is None: start_month = await fy.get_fiscal_year_start_month(db) if as_of_period is None: as_of_period = fy.default_as_of_period(fiscal_year, start_month) all_budgets = await list_budgets(db, fiscal_year) all_scope_conflicts, all_conflicted_budget_ids = diagnose_scope_conflicts( all_budgets, ) budgets = all_budgets if owner_person_id is not None: budgets = [b for b in budgets if b.owner_person_id == owner_person_id] visible_budget_ids = {budget.id for budget in budgets} conflicted_budget_ids = all_conflicted_budget_ids.intersection( visible_budget_ids, ) if owner_person_id is None: scope_conflicts = all_scope_conflicts else: # Owner/self-service HTTP and MCP reads must still fail closed when # an owned budget overlaps another owner's or company-wide budget, # without disclosing the hidden budget's identifier. scope_conflicts = [ { "budget_ids": [ budget_id for budget_id in conflict["budget_ids"] if budget_id in visible_budget_ids ], "reason": conflict["reason"], } for conflict in all_scope_conflicts if any( budget_id in visible_budget_ids for budget_id in conflict["budget_ids"] ) ] facts = await _actuals_for_year( db, fiscal_year, start_month, withhold_cloud=withhold_cloud, ) elapsed = fy.months_elapsed(fiscal_year, as_of_period, start_month) as_of_pfx = as_of_period fy_lo, fy_hi = fy.fy_bounds(fiscal_year, start_month) # Open-PO commitments (approved/received, not fully paid) — forward # spend exposure that hasn't hit actuals yet. Read via the governance # coordinator (cross-domain procurement read stays out of this domain). from app.core.coordinators import finance_governance_coordinator as gov commit_rows = (await gov.open_commitments(db)).get("commitments", []) items = [] tot_budget = Decimal("0") tot_ytd = Decimal("0") tot_forecast = Decimal("0") tot_committed = Decimal("0") for b in budgets: annual = _q2(b.amount_usd) ytd_actual = Decimal("0") month_actual = Decimal("0") for f in facts: if not _fact_matches(f, b): continue amt = _q2(f.amount_usd) # period 'YYYY-MM' <= as_of_period sorts lexically (same width). if f.period <= as_of_pfx: ytd_actual += amt if f.period == as_of_pfx: month_actual += amt ytd_budget = _q2(annual * Decimal(elapsed) / Decimal(12)) if elapsed else Decimal("0") month_budget = _q2(annual / Decimal(12)) run_rate = _q2(ytd_actual / Decimal(elapsed) * Decimal(12)) if elapsed else Decimal("0") # Open commitments matching this budget's scope. committed = Decimal("0") for c in commit_rows: if _commit_matches(c, b): committed += _q2(c.get("amount_usd")) # Remaining headroom AFTER actuals AND open commitments — the # working-capital number a budget owner can still spend. remaining = _q2(annual - ytd_actual - committed) # Commitment-aware forecast: year-end exposure can't be below what's # already spent + firmly committed, so floor the run-rate at it. forecast = max(run_rate, _q2(ytd_actual + committed)) burn_pct = round(float(ytd_actual / annual * 100), 1) if annual > 0 else 0.0 pace_pct = round(float(ytd_actual / ytd_budget * 100), 1) if ytd_budget > 0 else 0.0 commit_burn_pct = ( round(float((ytd_actual + committed) / annual * 100), 1) if annual > 0 else 0.0 ) # Status: forecast (committed-aware) will exceed budget → over; # pacing hot OR committed exposure already past budget → warning. if annual > 0 and forecast > annual: status = "over" elif pace_pct >= 100 or burn_pct >= 80 or commit_burn_pct >= 100: status = "warning" else: status = "ok" scope_conflict = b.id in conflicted_budget_ids # Historical overlap has no safe automatic winner: counting both # repeats the same facts, while choosing one silently changes the # approved plan. Keep both detail rows visible for remediation but # fail closed by excluding every conflicted budget from aggregate # totals and alert delivery. if not scope_conflict: tot_budget += annual tot_ytd += ytd_actual tot_forecast += forecast tot_committed += committed items.append({ "id": b.id, "scope_label": _scope_label(b), "cost_type": b.cost_type, "department": b.department, "cost_center": b.cost_center, "owner_person_id": b.owner_person_id, "annual_budget_usd": str(annual), "ytd_budget_usd": str(ytd_budget), "ytd_actual_usd": str(_q2(ytd_actual)), "month_budget_usd": str(month_budget), "month_actual_usd": str(_q2(month_actual)), "committed_usd": str(_q2(committed)), "remaining_usd": str(remaining), "variance_ytd_usd": str(_q2(ytd_budget - ytd_actual)), "forecast_year_end_usd": str(forecast), "projected_variance_usd": str(_q2(annual - forecast)), "burn_pct": burn_pct, "pace_pct": pace_pct, "commit_burn_pct": commit_burn_pct, "status": status, "scope_conflict": scope_conflict, }) items.sort(key=lambda r: ({"over": 0, "warning": 1, "ok": 2}[r["status"]], -float(r["ytd_actual_usd"]))) return { "fiscal_year": fiscal_year, "as_of_period": as_of_period, "fiscal_year_start_month": start_month, "fy_label": fy.fy_label(fiscal_year, start_month), "fy_period_start": fy_lo, "fy_period_end": fy_hi, "months_elapsed": elapsed, "budget_count": len(items), "valid_budget_count": len(items) - len(conflicted_budget_ids), "scope_conflict_count": len(scope_conflicts), "scope_conflicts": scope_conflicts, "totals_exclude_scope_conflicts": bool(scope_conflicts), "total_annual_budget_usd": str(_q2(tot_budget)), "total_ytd_actual_usd": str(_q2(tot_ytd)), "total_committed_usd": str(_q2(tot_committed)), "total_remaining_usd": str(_q2(tot_budget - tot_ytd - tot_committed)), "total_forecast_year_end_usd": str(_q2(tot_forecast)), "total_projected_variance_usd": str(_q2(tot_budget - tot_forecast)), "over_count": sum( 1 for item in items if not item["scope_conflict"] and item["status"] == "over" ), "warning_count": sum( 1 for item in items if not item["scope_conflict"] and item["status"] == "warning" ), "items": items, "withheld_cost_types": _withheld_meta(withhold_cloud), } # ─── Alerting (folded into finance.spend_rollup) ──────────────── def _budget_matches_alert_snapshot( budget: FinanceBudget, item: dict, ) -> bool: """True when the locked row still matches the report that triggered it.""" return ( str(_q2(budget.amount_usd)) == item["annual_budget_usd"] and budget.cost_type == item["cost_type"] and budget.department == item["department"] and budget.cost_center == item["cost_center"] and ( str(budget.owner_person_id) if budget.owner_person_id is not None else None ) == ( str(item["owner_person_id"]) if item["owner_person_id"] is not None else None ) ) async def evaluate_and_alert(db: AsyncSession, *, fiscal_year: int, as_of_period: str, start_month: int | None = None) -> dict: """Enqueue one de-duped burn alert per budget that crosses a threshold. Recipient = the budget owner only (a cost-centre owner / manager). No owner → no alert (never a mass employee push). Monotonic dedup via ``last_alerted_threshold`` — a crossing fires once and re-arms only if the ceiling is later raised. The pending outbox row and marker are committed together by the caller; external delivery belongs exclusively to the claim-first outbox worker and never runs in this transaction. """ from app.identity.person_models import IdentityPerson from app.notifications.service import ( DELIVERY_ROLE_GUARD_CONTEXT_KEY, CardStat, LocalizedText, NotificationEnvelope, NotificationService, ) report = await budget_vs_actual(db, fiscal_year, as_of_period, start_month=start_month) fy_text = report["fy_label"] svc = NotificationService(db) enqueued = 0 # Load the stored roles for every potentially confidential recipient in # one query. Alert evaluation can cover hundreds of cost centres; doing # this inside the loop would add a new N+1 to the daily rollup. confidential_owner_ids = { item["owner_person_id"] for item in report["items"] if item["owner_person_id"] and item["cost_type"] not in _NON_CLOUD_COST_TYPES } role_names_by_owner: dict[object, list[str]] | None = {} if confidential_owner_ids: from app.auth.models import AuthRole, AuthUserRole async with best_effort( db, label="budget_service best-effort step", log=logger, ) as be_706_step: loaded_role_names: dict[object, list[str]] = {} role_rows = (await db.execute( select(AuthUserRole.user_id, AuthRole.name) .join(AuthRole, AuthRole.id == AuthUserRole.role_id) .where(AuthUserRole.user_id.in_(confidential_owner_ids)) )).all() for role_owner_id, role_name in role_rows: loaded_role_names.setdefault(role_owner_id, []).append(role_name) role_names_by_owner = loaded_role_names if not be_706_step.ok: # Non-cloud owners remain deliverable, but every cloud/mixed # recipient fails closed below rather than inheriting money detail. logger.error("finance.budget: confidential owner role lookup failed") role_names_by_owner = None for item in report["items"]: if item.get("scope_conflict"): # Existing corrupt inventory remains visible in the report but # cannot emit two alerts for the same underlying actual. continue # Trigger on the YTD-PRORATED pace (ytd_actual / ytd_prorated_budget), # not the raw annual burn (ytd_actual / annual). The docstring and the # dashboard's ``warning`` status both key on prorated pacing; keying # the alert on annual burn instead made it fire far too late (a budget # pacing at 150% of plan in March has only ~12% annual burn and would # not alert until ~80% of the WHOLE year was consumed — past the point # the owner can act). ``burn`` stays as secondary context in the card. pace = item["pace_pct"] burn = item["burn_pct"] crossed = next((t for t in _ALERT_THRESHOLDS if pace >= t), 0) owner_id = item["owner_person_id"] if not crossed or not owner_id: continue # Serialize the threshold claim. Two schedulers may evaluate the same # report concurrently; the second must wait for the first transaction # and then observe its advanced marker instead of enqueuing a twin. budget = (await db.execute( select(FinanceBudget) .where(FinanceBudget.id == item["id"]) .with_for_update() .execution_options(populate_existing=True) )).scalar_one_or_none() if budget is None: continue # The report was built before this lock. An admin may have changed the # amount, matching scope, or owner while we waited. Never render/send # from that stale snapshot (especially to a former owner); the next # scheduled run will recompute against the now-committed row. if not _budget_matches_alert_snapshot(budget, item): logger.info( "finance.budget: skip stale alert snapshot for budget %s", item["id"], ) continue if budget.last_alerted_threshold >= crossed: continue owner = await db.get(IdentityPerson, owner_id) if owner is None or owner.employment_status != "active": # Still advance the marker so a missing owner doesn't re-fire daily. budget.last_alerted_threshold = crossed continue severity = "已超预算" if crossed >= 100 else "预算预警" try: # #2281: budget owners are arbitrary people, not necessarily # finance users. A cloud budget reads invoice-net actuals, and a # wildcard (cost_type=NULL) budget may contain those same rows. # Determine the cron recipient's STORED roles for each envelope; # there is no request-scoped X-View-Role in a scheduled job. # # Explicitly non-cloud budgets keep their full, useful detail. # Cloud/wildcard/unknown budgets default to the amount-free twin # unless the owner is positively proven to be in the invoice circle. # If this role query fails, the exception is caught below and no # envelope is dispatched: failure cannot fall through to the # confidential rendering. redact_cloud_money = False delivery_guard: dict | None = None if item["cost_type"] not in _NON_CLOUD_COST_TYPES: from app.core.rbac import ( CLOUD_ADMIN_ROLE, FINANCE_ROLE, GLOBAL_ADMIN_ROLE, can_view_invoice_measure, ) if role_names_by_owner is None: raise RuntimeError("confidential owner role lookup unavailable") role_names = role_names_by_owner.get(owner.id, []) redact_cloud_money = not can_view_invoice_measure(role_names) safe_body_text = ( f"{item['scope_label']} {fy_text} 预算使用进度已达到" f"{severity}条件。本通知按云实付金额保密策略不展示" "实际、预测、偏差或可反推数据。请到 Helix 财务系统 " "→ 预算 复核，或联系财务 / 云管理员协助核对。" ) safe_subtitle = "预算达到预警条件 · 云实付金额已隐藏" delivery_guard = { "any_roles": [ FINANCE_ROLE, CLOUD_ADMIN_ROLE, GLOBAL_ADMIN_ROLE, ], "fallback": { "title": f"[{severity}] {item['scope_label']} {fy_text} 预算", "body_text": safe_body_text, "url": "/me/budget", "card_payload": {"subtitle": safe_subtitle}, }, } if redact_cloud_money: # Do not include an absolute amount OR a percentage paired # with the known annual budget. Either combination would # reveal or reconstruct invoice-net cloud spend. The alert's # actionable signal survives as a categorical threshold # crossing, with a link for follow-up by the finance/cloud # circle. body_text = safe_body_text card_subtitle = safe_subtitle card_stats: list[CardStat] = [] else: body_text = ( f"{item['scope_label']} {fy_text} 年度预算 " f"${item['annual_budget_usd']} USD,截至 {as_of_period} " f"已用 ${item['ytd_actual_usd']}(为同期计划进度的 {pace}%、" f"年累 {burn}%),按当前速率预计年底 " f"${item['forecast_year_end_usd']}。" f"请到 Helix 财务系统 → 预算 复核。" ) card_subtitle = f"同期进度 {pace}% · 阈值 {crossed}%" overspend = crossed >= 100 card_stats = [ CardStat(label=LocalizedText(zh="同期进度", en="Pace vs plan", ja="計画比の進捗"), value=f"{pace}%", trend=f"年累 {burn}%", trend_is_good=False, hint=LocalizedText( zh=f"已用 ${item['ytd_actual_usd']} / 年度预算 " f"${item['annual_budget_usd']}", en=f"Spent ${item['ytd_actual_usd']} / annual " f"budget ${item['annual_budget_usd']}", ja=f"執行済み ${item['ytd_actual_usd']} / 年間予算 " f"${item['annual_budget_usd']}", )), CardStat(label=LocalizedText(zh="年底预测", en="Year-end forecast", ja="年度末予測"), value=f"${item['forecast_year_end_usd']}", hint=LocalizedText( zh=f"预计{'超支' if overspend else '偏差'} " f"${item['projected_variance_usd']}", en=f"Projected " f"{'overspend' if overspend else 'variance'} " f"${item['projected_variance_usd']}", ja=f"{'超過' if overspend else '差異'}見込み " f"${item['projected_variance_usd']}", )), ] await svc.enqueue(NotificationEnvelope( recipient=owner, title=f"[{severity}] {item['scope_label']} {fy_text} 预算", body_text=body_text, channels=["feishu"], # Owners can be ordinary employees; route every alert to the # owner-scoped page instead of the role-gated admin console. url="/me/budget", card_subtitle=card_subtitle, card_stats=card_stats, context=( {DELIVERY_ROLE_GUARD_CONTEXT_KEY: delivery_guard} if delivery_guard is not None else {} ), related_resource_type="finance.budget", )) budget.last_alerted_threshold = crossed enqueued += 1 except Exception: logger.exception("finance.budget: alert enqueue failed for budget %s", item["id"]) return {"checked": report["budget_count"], "alerts_enqueued": enqueued}