"""Software monthly billing — close / reopen / list / preview. See migration 063 and ``docs/ops/software-billing-ops.md`` for the full design rationale. Summary: * A "period" is ``YYYY-MM``. The current month can be previewed live but not closed until it has ended. Closing an already-closed period is a no-op. Reopen deletes the period's rows so the next close regenerates them — this is the supported correction flow. * A seat allocation is included in a period's bill if it was active at any point during the period (``allocated_at <= period_end`` and (``revoked_at is NULL`` or ``revoked_at >= period_start``)). * Amount per period is derived from ``billing_cycle``: - ``monthly`` → full ``plan.price`` - ``yearly`` → ``plan.price / 12`` (spread across every active month) - ``one_time`` → full ``plan.price`` only in the allocation's month, else zero. Zero-amount rows are omitted to keep the table lean. """ from __future__ import annotations import asyncio import re import uuid from datetime import UTC, date, datetime from decimal import ROUND_HALF_UP, Decimal from typing import Any from app.core.safe_logging import get_safe_logger logger = get_safe_logger(__name__) from sqlalchemy import and_, delete, func, or_, select, text from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.core.exceptions import AppException from app.domains.asset.models import ( SoftwareLicensePlan, SoftwareMonthlyBill, SoftwareMonthlyBillHistory, SoftwareSeatAllocation, SoftwareSeatSyncStat, ) from app.identity.person_models import IdentityPerson _PERIOD_FMT = "%Y-%m" _Q = Decimal("0.01") # Strict match — strptime("%Y-%m") accepts single-digit months like # "2026-1", which would then produce period="2026-01" internally and # silently mismatch stored periods. Require the canonical form. _PERIOD_RE = re.compile(r"^\d{4}-(0[1-9]|1[0-2])$") def _parse_period(period: str) -> tuple[date, date]: """Return (period_start, period_end_exclusive). Raises AppException if malformed. ``period_end_exclusive`` is the first day of the following month — inclusive tests use ``< end``.""" if not _PERIOD_RE.match(period): raise AppException( code="INVALID_PERIOD", message=f"Period must be YYYY-MM with two-digit month, got {period!r}", status_code=400, ) try: start = datetime.strptime(period, _PERIOD_FMT).date().replace(day=1) except ValueError as exc: raise AppException( code="INVALID_PERIOD", message=f"Period must be YYYY-MM, got {period!r}", status_code=400, ) from exc # First day of next month — avoids leap-year / 31-vs-30 branching. if start.month == 12: end = date(start.year + 1, 1, 1) else: end = date(start.year, start.month + 1, 1) return start, end def _current_period() -> str: """Current YYYY-MM period in the business timezone (Asia/Shanghai). Anchored to Asia/Shanghai because that is the company's billing cadence — a "month" closes when the China business day rolls over, not when UTC midnight hits. The auto-close cron is also seeded in Asia/Shanghai (``"3 3 1 * *"`` in migration 064); using UTC here would leave a 16-hour window every month-start where the cron fires but the period still reads "current" → ``PERIOD_NOT_ENDED`` skip → no auto-close ever happens, which is the failure mode the pre-launch logic review caught (2026-05-03). """ from zoneinfo import ZoneInfo return datetime.now(ZoneInfo("Asia/Shanghai")).strftime(_PERIOD_FMT) def _is_past_period(period: str) -> bool: return period < _current_period() def _quantize(value: Decimal) -> Decimal: return value.quantize(_Q, rounding=ROUND_HALF_UP) def _compute_amount( *, price: Decimal, billing_cycle: str, period_start: date, period_end_excl: date, allocated_at: datetime, ) -> Decimal: """Per-period amount for one allocation. See module docstring.""" if billing_cycle == "monthly": return _quantize(price) if billing_cycle == "yearly": return _quantize(price / Decimal(12)) if billing_cycle == "one_time": # Only charge in the month where the seat was first allocated. # Normalise to UTC first so a timestamp near month boundaries # is bucketised consistently regardless of the originating tz. if allocated_at.tzinfo is not None: alloc_day = allocated_at.astimezone(UTC).date() else: alloc_day = allocated_at.date() if period_start <= alloc_day < period_end_excl: return _quantize(price) return Decimal("0.00") # Unknown cycle — log + return 0 so the row drops out of the # bill instead of being silently over-charged. The previous shape # treated unknown as monthly, which produces a 3×/12× over-bill # if a future migration introduces ``quarterly`` / ``annual`` # without teaching the billing code about it (P1 from CR1). # The 0 amount surfaces as a missing line in the dashboard, # which is exactly the signal we want admin to notice and fix. get_safe_logger(__name__).error( "billing_service._compute_amount: unknown billing_cycle %r — " "row will be billed at 0; teach _compute_amount before " "introducing this cycle.", billing_cycle, ) return Decimal("0.00") def period_overlap_clauses(period_start: date, period_end_excl: date) -> list: """SQLAlchemy WHERE clauses selecting every ``SoftwareSeatAllocation`` that was billable at any point during ``[period_start, period_end_excl)``. Returned as a list so the caller AND-combines them via ``.where(*clauses)``. This is the SINGLE source of truth for "which seats bill in a period", shared by the close/preview path (``_fetch_allocations_overlapping``) AND the live "当月(实时)" cost-stats view in ``asset.router``. The two surfaces MUST agree on the billable set — they only legitimately differ on price (live reads the current plan price; a closed snapshot freezes it). Keeping the predicate here prevents the historical bug where the live view filtered ``status == 'active'`` and silently dropped seats cancelled / tier-changed mid-month that the bill still charged. Three clauses, in order: 1. ``allocated_at < period_end_excl`` — the seat was created before the period ended. 2. ``status IN ('active', 'revoked')`` — pending allocations are NOT billable. Pre-launch hardening 2026-05-05: the previous shape had no status filter, so a request that was rejected/cancelled by Feishu but whose row stayed in ``status='pending'`` (the original callback only flipped ``employee_request.status``, not the allocation's) still got billed prorated for the period — directly violating red-line #3 (历史/账单准确性). The companion fix in the Feishu callback now sets ``allocation.status='revoked'`` on rejection; migration 119 backfills any stuck rows. 3. ``revoked_at IS NULL OR (revoked_at > allocated_at AND revoked_at > period_start)`` — non-revoked, OR revoked with a non-zero billable window that overlaps the period. The ``revoked_at > allocated_at`` clause excludes rejected rows whose ``revoked_at`` was zero-aligned to ``allocated_at`` (so they never hit a bill at all). The ``revoked_at > period_start`` bound is STRICT (not ``>=``): a seat whose access ends exactly at the first instant of a period had zero billable time IN that period, so it must not be charged. This is what lets a self-service "cancel effective end-of-month" (auto-revoke stamps ``revoked_at`` = start of the next month) drop cleanly out of that next month's bill instead of being charged a full extra un-prorated month. """ start_dt = datetime.combine(period_start, datetime.min.time(), tzinfo=UTC) end_dt = datetime.combine(period_end_excl, datetime.min.time(), tzinfo=UTC) return [ SoftwareSeatAllocation.allocated_at < end_dt, SoftwareSeatAllocation.status.in_(("active", "revoked")), or_( SoftwareSeatAllocation.revoked_at.is_(None), and_( SoftwareSeatAllocation.revoked_at > start_dt, SoftwareSeatAllocation.revoked_at > SoftwareSeatAllocation.allocated_at, ), ), ] async def _fetch_allocations_overlapping( db: AsyncSession, period_start: date, period_end_excl: date, ) -> list[SoftwareSeatAllocation]: """Every allocation active at any point during the period. Selection predicate lives in :func:`period_overlap_clauses` so the live cost-stats view can reuse the exact same logic — see that function's docstring for the per-clause rationale. """ stmt = ( select(SoftwareSeatAllocation) .options( selectinload(SoftwareSeatAllocation.plan).selectinload( SoftwareLicensePlan.software, ), ) .where(*period_overlap_clauses(period_start, period_end_excl)) # Deterministic order makes close → reopen → close diffs obvious # to a reviewer comparing rows. .order_by(SoftwareSeatAllocation.allocated_at.asc()) ) return list((await db.execute(stmt)).scalars().all()) async def _build_lines_for_period( db: AsyncSession, period: str, ) -> list[dict[str, Any]]: """Compute (not persist) the bill lines for a period. Shared by the live preview and the close path — same computation, different sink.""" period_start, period_end_excl = _parse_period(period) allocations = await _fetch_allocations_overlapping(db, period_start, period_end_excl) # NOTE: do NOT early-return when there are no seat allocations — a period # can have zero seats but still carry one-time add-on charges, and those # must still be billed. The addon lines are appended below regardless. user_ids = {a.user_id for a in allocations} users: dict[uuid.UUID, IdentityPerson] = {} if user_ids: result = ( await db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(user_ids)) ) ).scalars().all() users = {u.id: u for u in result} lines: list[dict[str, Any]] = [] for a in allocations: plan = a.plan sw = plan.software if plan else None if plan is None: # Orphaned allocation — skip. Shouldn't happen; the FK # prevents it. Defensive in case of DB-level drift. continue amount = _compute_amount( price=plan.price, billing_cycle=plan.billing_cycle, period_start=period_start, period_end_excl=period_end_excl, allocated_at=a.allocated_at, ) # Zero-amount rows are noise in the finalised table (e.g. # one_time seats that aren't in their allocation month). Skip. if amount == Decimal("0.00"): continue user = users.get(a.user_id) # As-of-period-end status: if revoked before period end, the # seat is historically "revoked" in this bill. Otherwise active. end_dt = datetime.combine(period_end_excl, datetime.min.time(), tzinfo=UTC) # ``revoked_at`` is declared ``DateTime(timezone=True)`` on the # ORM model, but MySQL's native DATETIME doesn't store a # timezone, so SQLAlchemy returns a naive datetime when the # value was written without one (older revokes before the # ``UTC.now()`` convention was uniformly applied). Compare # against ``end_dt`` only after coercing the value to UTC-aware, # otherwise Python raises # "can't compare offset-naive and offset-aware datetimes" and # the whole bill load 500s — see 2026-05-01 incident. revoked = a.revoked_at if revoked is not None and revoked.tzinfo is None: revoked = revoked.replace(tzinfo=UTC) if revoked is not None and revoked < end_dt: line_status = "revoked" else: line_status = "active" # Currency is plan-level. Default to ``CNY`` for legacy plans # whose row predates the column (server_default backfills new # rows; this guard handles in-memory ORM objects loaded before # the migration ran in old test fixtures). # ``plan.currency`` is NOT NULL with default 'CNY' in # migration 116; treat it as always present. Direct access # surfaces wrong-column-name bugs immediately instead of # silently masking with the legacy fallback (CLAUDE.md # ``feedback_no_getattr_default_for_model_columns``). plan_currency = (plan.currency or "CNY").upper() lines.append({ "line_type": "seat", "charge_id": None, "quantity": None, "allocation_id": a.id, "software_id": sw.id if sw else None, "software_name": sw.name if sw else None, "plan_id": plan.id, "plan_name": plan.name, "user_id": a.user_id, "user_name": user.display_name if user else None, "user_email": user.work_email if user else None, "user_employee_no": user.employee_no if user else None, "user_employment_status": user.employment_status if user else None, "department": user.department if user else None, # Top-level BU (org_path first segment, tenant root stripped) # so the cost view can roll leaf departments up to their # business unit. ``department_chain`` is leaf-last minus the # tenant root, so ``[0]`` is the BU. "bu": (user.department_chain[0] if user and user.department_chain else None), "reason": a.reason, "price": plan.price, "billing_cycle": plan.billing_cycle, "currency": plan_currency, "amount": amount, "allocated_at": a.allocated_at, "revoked_at": a.revoked_at, "status": line_status, }) # One-time add-on charges approved against this period — materialised # as ``addon`` lines so every cost rollup that SUMs this period picks # them up identically to seats (the "钱不丢" guarantee). lines.extend(await _build_addon_lines_for_period(db, period)) return lines async def _build_addon_lines_for_period( db: AsyncSession, period: str, ) -> list[dict[str, Any]]: """Bill lines for one-time add-on charges pinned to ``period``. An add-on charge belongs to exactly one period — the open period it was approved in — so selection is a simple ``status='approved' AND period=?`` rather than the seat path's interval overlap. The ``amount`` is already frozen on the charge (unit_price × quantity at request time); the department snapshot was taken at approval. """ from app.domains.asset.models import ( AssetSoftwareCatalog, SoftwareAddonCharge, SoftwareAddonItem, ) rows = (await db.execute( select(SoftwareAddonCharge, SoftwareAddonItem, AssetSoftwareCatalog) .join(SoftwareAddonItem, SoftwareAddonCharge.item_id == SoftwareAddonItem.id) .outerjoin( AssetSoftwareCatalog, SoftwareAddonCharge.software_id == AssetSoftwareCatalog.id, ) .where( SoftwareAddonCharge.status == "approved", SoftwareAddonCharge.period == period, ) .order_by(SoftwareAddonCharge.approved_at.asc()) )).all() if not rows: return [] user_ids = {c.user_id for c, _, _ in rows} users: dict[uuid.UUID, IdentityPerson] = {} if user_ids: result = (await db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(user_ids)) )).scalars().all() users = {u.id: u for u in result} lines: list[dict[str, Any]] = [] for charge, item, sw in rows: amount = charge.amount if amount is None or amount == Decimal("0.00"): continue user = users.get(charge.user_id) # Department is the snapshot frozen at approval (attribution key); # fall back to the live person if a legacy charge lacks it. BU is # derived live for the cost view's roll-up only. department = charge.department or (user.department if user else None) lines.append({ "line_type": "addon", "charge_id": charge.id, "quantity": charge.quantity, "allocation_id": None, "software_id": charge.software_id, "software_name": sw.name if sw else None, "plan_id": None, # The add-on item name (e.g. "AI Credits") lands in plan_name # so existing per-line displays render a meaningful label. "plan_name": item.name if item else None, "user_id": charge.user_id, "user_name": user.display_name if user else None, "user_email": user.work_email if user else None, "user_employee_no": user.employee_no if user else None, "user_employment_status": user.employment_status if user else None, "department": department, "bu": (user.department_chain[0] if user and user.department_chain else None), "reason": charge.reason, "price": charge.unit_price, "billing_cycle": "one_time", "currency": (charge.currency or "CNY").upper(), "amount": amount, "allocated_at": charge.approved_at or charge.created_at, "revoked_at": None, "status": "active", }) return lines def _stat_family_key(product_key: str | None, plan_id: uuid.UUID) -> str: """Dedup key for a ``software_seat_sync_stat`` row. ``vendor_license_holders`` is a site-scoped count of a vendor product *family* keyed by ``product_key``, written once per *license plan*. Multiple stat rows on one software that share a ``product_key`` (e.g. a Jira family modelled as several same-site add-on catalogs) carry the same family count, so we collapse them to ONE representative via ``max`` — the ``max`` is **load-bearing**, not cosmetic: it prevents same-family add-on rows from being summed (which would re-introduce the pre-#593 double-count). The key falls back to the plan id when ``product_key`` is NULL (e.g. Slack, in-house catalogs) so distinct unkeyed plans never collapse together. Note (2026-05-31): this key is INTENTIONALLY site-agnostic. Two genuinely independent same-product subscriptions on disjoint sites (e.g. Confluence on ``tidb`` vs an independent ``pingcap-us`` org) would wrongly ``max``-collapse here. We do NOT model that by changing the key; instead each subscription is a distinct *catalog/software* with its own contracted ``max_seats``, and a subscription with no contract on file is left inactive (its stat row removed) rather than counted from raw product_access. If a future independent paid org is fully contracted, move this key to ``(product_key, frozenset(paid_sites))`` so disjoint- site subscriptions sum while same-site add-ons still ``max``-dedup. """ return product_key if product_key else f"__plan__:{plan_id}" def _effective_seats( vendor_license_holders: int | None, max_seats: int | None, ) -> int: """Billable seat count for one plan = the lesser of vendor access holders and the contracted ceiling (``max_seats``). ``vendor_license_holders`` counts accounts the vendor's Org-Admin API reports as *holding* the product (product_access). For most products (Jira/JSM) that equals the billed seat count and is ``<= max_seats`` so the clamp is a no-op (Jira prod: 372 holders ≤ 400 contracted). But some products grant broad *access* that does not consume a paid license — observed on Atlassian Confluence prod 2026-05: ~100 active product_access holders against a 10-seat purchased subscription (the renewal quote bills 10, not 100). Reporting the raw access count as "生效坐席" overstates paid seats ~10x, so we clamp to the contracted ceiling, which is the authoritative paid-seat number. The clamped-away excess (``vendor_license_holders > max_seats``) is not lost: callers surface it as an ``under_licensed`` finding (active access exceeds licensed seats) — the governance signal IT actually wants. ``max_seats=None`` means no contract ceiling on file (non-Atlassian catalogs that count by distinct users), so we return the raw holder count unchanged. """ vlh = int(vendor_license_holders or 0) if max_seats is None: return vlh return min(vlh, int(max_seats)) async def _vendor_seats_by_software( db: AsyncSession, software_ids: set[uuid.UUID], ) -> dict[str, dict[str, int]]: """Per-software ``vendor_license_holders`` snapshot from the latest ``*.seat_sync`` run, **deduplicated by product family** within each software. This is the AUTHORITATIVE seat count — it matches what the vendor's admin console shows ("Atlassian 72/100"). The billing-line count (``len(rows)``) is NOT a substitute: tier-changes and mid-month re-grants inflate the line count to a multiple of the real seat count (JSM observed 175 lines for 72 actual seats). The workbench surfaces THIS number under "生效坐席" so the cell agrees with the drill-down drawer's "厂商许可" stat. ``vendor_license_holders`` is family-keyed (see :func:`_stat_family_key`), so naively ``SUM``-ing it over plans double-counts whenever two active plans on one software share a ``product_key`` (allowed — ``SoftwareLicensePlan`` has no per-software uniqueness). We collapse to ONE representative value per ``product_key`` (``max``; see :func:`_stat_family_key` for why ``max`` is load-bearing). Each plan's contribution is first clamped to its contracted ceiling via :func:`_effective_seats` — the raw product_access holder count overstates paid seats for products with broad non-licensing access (Confluence). The clamp is a no-op for healthy products (holders ≤ ``max_seats``). Returns ``{software_id_str: {family_key: vendor_seat_count}}`` so both the per-software flat total (:func:`_flatten_vendor_seats`) and the cross-software family-deduped top stat (:func:`_seat_count_from_lines`) can be derived without re-querying. Empty when no software has been seat-synced (in-house catalog entries / new integrations) — callers fall back to distinct-user-count from the billable set. """ if not software_ids: return {} rows = ( await db.execute( select( SoftwareSeatSyncStat.software_id, SoftwareSeatSyncStat.product_key, SoftwareSeatSyncStat.plan_id, SoftwareSeatSyncStat.vendor_license_holders, SoftwareLicensePlan.max_seats, ) .join( SoftwareLicensePlan, SoftwareLicensePlan.id == SoftwareSeatSyncStat.plan_id, ) .where(SoftwareSeatSyncStat.software_id.in_(software_ids)) ) ).all() out: dict[str, dict[str, int]] = {} for sw_id, product_key, plan_id, vlh, max_seats in rows: fam = out.setdefault(str(sw_id), {}) fkey = _stat_family_key(product_key, plan_id) fam[fkey] = max(fam.get(fkey, 0), _effective_seats(vlh, max_seats)) return out def _flatten_vendor_seats( vendor_seats_by_software: dict[str, dict[str, int]], ) -> dict[str, int]: """Collapse the nested ``{software: {family: count}}`` structure to the flat ``{software_id_str: vendor_seat_total}`` shape the API response (``vendor_seats_by_software``) and the frontend per-row column expect. Per software, the total is the SUM over DISTINCT product families — already deduped by :func:`_vendor_seats_by_software`, so a software carrying two plans of the same family contributes that family once. """ return { sw_key: sum(fams.values()) for sw_key, fams in vendor_seats_by_software.items() } def _seat_count_from_lines( lines: list[dict[str, Any]], vendor_seats_by_software: dict[str, dict[str, int]], ) -> int: """Top-level "生效坐席" — the seat-count statistic that aggregates across every software in the response. For each software: * If the seat-sync table has a vendor count, use it (authoritative). * Otherwise (catalog entry never synced) fall back to the distinct ``user_id`` count among that software's billable rows — one person using a software once = one seat, regardless of tier-changes or re-grants in the same period. Crucially, vendor counts are deduplicated by product family ACROSS software before summing: a family (e.g. ``jira-software``) modelled as several catalog rows (Jira Software + the Tempo / ScriptRunner / eazyBI marketplace-addon catalogs, all carrying the identical org-wide 364) or one product split per site (Confluence tidb + Confluence US) must contribute its seat count ONCE, not once per catalog row — otherwise the top card reads a multiple of the real headcount (prod 2026-05-30: 3241 shown vs 2119 real, a 1122-seat / 53% inflation). Lines without a ``software_id`` fall into a single "unknown" bucket counted via distinct users. The sum is what the workbench shows in the top stat card. """ by_sw: dict[str, set[uuid.UUID]] = {} for line in lines: # One-time add-on charges are NOT seats — they have no vendor # license holder. Counting their user_id here would inflate the # 生效坐席 stat (and the preview's ``active_seats``) for any # software whose add-on charges land in a period. Skip them. if line.get("line_type") == "addon": continue sw = line.get("software_id") key = str(sw) if sw is not None else "" bucket = by_sw.setdefault(key, set()) uid = line.get("user_id") if uid is not None: bucket.add(uid) total = 0 # Vendor-counted side: dedup families across all software that have a # seat-sync row. Same family on multiple software rows → counted once. family_seats: dict[str, int] = {} for fams in vendor_seats_by_software.values(): for fkey, count in fams.items(): family_seats[fkey] = max(family_seats.get(fkey, 0), count) total += sum(family_seats.values()) # Fallback side: software with no vendor sync row (and the unknown # bucket) contribute their in-scope distinct-user count. for sw_key, users in by_sw.items(): if sw_key and sw_key in vendor_seats_by_software: continue total += len(users) return total def _aggregate_by_currency( lines: list[dict[str, Any]], ) -> tuple[dict[str, Decimal], Decimal | None, bool]: """Group line ``amount`` by ``currency``. Returns ``(by_currency, single_total_or_none, currency_warning)``: * ``by_currency`` — ``{currency: total}`` quantised. Always present even when empty so callers don't have to None-guard. * ``single_total_or_none`` — when exactly one currency is present, the same total as ``by_currency``'s only value (kept for back-compat with the legacy single-decimal callers); ``None`` when the lines mix multiple currencies (caller must read ``by_currency``). * ``currency_warning`` — ``True`` when ≥ 2 currencies appear. """ by_cur: dict[str, Decimal] = {} for line in lines: cur = (line.get("currency") or "CNY").upper() amt = line["amount"] if not isinstance(amt, Decimal): amt = Decimal(str(amt)) by_cur[cur] = by_cur.get(cur, Decimal("0.00")) + amt by_cur = {k: _quantize(v) for k, v in by_cur.items()} if len(by_cur) <= 1: single = next(iter(by_cur.values()), Decimal("0.00")) return by_cur, single, False return by_cur, None, True async def get_period_status(db: AsyncSession, period: str) -> str: """One of: * ``finalized`` — rows already exist * ``draft`` — current or future month, previewable but not closable * ``available`` — past month with no rows yet, closable """ _parse_period(period) # validate count = ( await db.execute( select(func.count(SoftwareMonthlyBill.id)).where( SoftwareMonthlyBill.period == period ) ) ).scalar_one() if count > 0: return "finalized" if _is_past_period(period): return "available" return "draft" async def preview_current_or_draft( db: AsyncSession, period: str, ) -> dict[str, Any]: """Live preview for a not-yet-closed period. Always recomputed — never reads from ``software_monthly_bills``. Safe for any period (past or current); caller decides whether to offer close action. Multi-currency: ``total_amount`` is populated only when every line is in the same currency; otherwise it's ``None`` and callers must read ``by_currency``. ``currency_warning=True`` indicates at least two distinct currencies — the UI surfaces a "split required" hint rather than silently mixing decimals.""" lines = await _build_lines_for_period(db, period) by_currency, single_total, warn = _aggregate_by_currency(lines) # "生效坐席" = vendor-authoritative seat count, NOT bill-line count. # PR #579 (briefly) made this ``len(lines)`` so the seat number would # equal the bill receipt's line count, but for multi-tier products # (JSM Service Agent / Reporter / External; Confluence per-site) one # employee occupies multiple billable lines AND mid-month tier swaps # create extra revoked-then-active line pairs, so the count inflated # to a multiple of the real headcount (observed 175 lines vs 72 # actual Atlassian seats). The drill-down drawer shows the vendor's # 72 — the workbench must agree. ``_vendor_seats_by_software`` is # the source of truth (from ``software_seat_sync_stat``, populated # by every ``*.seat_sync`` cron). Software with no sync row falls # back to distinct ``user_id`` count of its billable rows. software_ids = {line["software_id"] for line in lines if line.get("software_id")} vendor_seats = await _vendor_seats_by_software(db, software_ids) active_seats = _seat_count_from_lines(lines, vendor_seats) return { "period": period, "status": "draft", "total_amount": single_total, "by_currency": {k: str(v) for k, v in by_currency.items()}, "total_currency_count": len(by_currency), "currency_warning": warn, "active_seats": active_seats, "vendor_seats_by_software": _flatten_vendor_seats(vendor_seats), "finalized_at": None, "finalized_by": None, "items": [{"period": period, "id": None, "finalized_at": None, **line} for line in lines], } async def close_month( db: AsyncSession, period: str, *, actor_id: uuid.UUID | None = None, ) -> dict[str, Any]: """Finalise the period. Refuses future / current months; refuses already-closed periods (caller should reopen first if correcting). Single commit — either all rows land or none. ``actor_id=None`` means "closed by the system" (the scheduled auto-close job); ``finalized_by`` stays NULL in that case, which the UI renders as "system" in the audit column.""" _parse_period(period) # validate format first if not _is_past_period(period): raise AppException( code="PERIOD_NOT_ENDED", message=f"Cannot close {period} — period has not ended yet.", status_code=400, ) # Per-period MySQL/TiDB advisory lock — prevents two admins (or an # admin + scheduled close) from racing on the same period and # double-inserting bill rows. Acquire with a short timeout (5s) so # a benign admin-vs-cron overlap doesn't immediately 409 the # second caller — the first finishes in <1s typically and the # second's existing-row check then naturally refuses with # PERIOD_ALREADY_CLOSED. Lock auto-releases when the session ends; # we ALSO release explicitly in the ``finally`` block below so a # mid-flight raise (FK violation, insert crash, etc.) doesn't leave # the lock orphaned on a pooled connection that may be reused for # an unrelated request. Lock name length capped at 64 chars # (MySQL constraint); period is "YYYY-MM" so the full name stays # well under that. lock_name = f"helix.billing.close.{period}" lock_acquired = ( await db.execute( text("SELECT GET_LOCK(:name, 5)"), {"name": lock_name} ) ).scalar() if lock_acquired is None: # NULL = MySQL advisory-lock subsystem error (not just busy). # Round-3 logic-scan P1. raise AppException( code="PERIOD_LOCK_INFRASTRUCTURE_ERROR", message=( f"Failed to acquire close lock for {period}: MySQL " "advisory-lock subsystem returned NULL. Retry; if " "persistent, escalate." ), status_code=503, ) if lock_acquired != 1: raise AppException( code="PERIOD_CLOSE_IN_PROGRESS", message=f"Period {period} close is already running in another session. Try again shortly.", status_code=409, ) try: existing = ( await db.execute( select(func.count(SoftwareMonthlyBill.id)).where( SoftwareMonthlyBill.period == period ) ) ).scalar_one() if existing and existing > 0: raise AppException( code="PERIOD_ALREADY_CLOSED", message=f"Period {period} is already closed. Reopen it first to regenerate.", status_code=409, ) lines = await _build_lines_for_period(db, period) now = datetime.now(UTC) for line in lines: db.add( SoftwareMonthlyBill( id=uuid.uuid4(), period=period, line_type=line.get("line_type", "seat"), allocation_id=line["allocation_id"], software_id=line["software_id"], plan_id=line["plan_id"], charge_id=line.get("charge_id"), quantity=line.get("quantity"), user_id=line["user_id"], software_name=line["software_name"], plan_name=line["plan_name"], user_name=line["user_name"], department=line["department"], reason=line["reason"], price=line["price"], billing_cycle=line["billing_cycle"], currency=line.get("currency", "CNY"), amount=line["amount"], allocated_at=line["allocated_at"], revoked_at=line["revoked_at"], finalized_at=now, finalized_by=actor_id, ) ) await db.flush() # COMMIT before the finally releases the advisory lock. The lock # only serialises callers while held — releasing it with rows # still uncommitted re-opened the race it exists to close: caller # B grabs the freed lock, its existing-row count runs in a fresh # REPEATABLE READ snapshot that cannot see A's uncommitted rows, # and B inserts the whole period again. Seat lines then die on # uq_(period, allocation_id); addon lines (allocation_id=NULL, # which MySQL UNIQUE treats as distinct) sailed through → real # double-billed spend in every downstream SUM(amount). Callers' # own db.commit() afterwards becomes a harmless no-op. await db.commit() by_currency, single_total, warn = _aggregate_by_currency(lines) return { "period": period, "line_count": len(lines), "total_amount": single_total, "by_currency": {k: str(v) for k, v in by_currency.items()}, "total_currency_count": len(by_currency), "currency_warning": warn, "finalized_at": now, } finally: # Release on every exit path — success OR raise — so an INSERT # failure mid-flight doesn't leave the advisory lock orphaned # on a pooled connection that the next request might reuse. # Inner try/except so a RELEASE_LOCK network hiccup never # masks the original exception from the try body (it would # propagate from ``finally`` instead and the caller would see # a useless lock-release error rather than the real failure). # Session-end auto-release is the safety net for the rare case # where this log line is all we get. # RELEASE_LOCK runs in a ``finally`` while the original exception may # still be propagating. Opening a SAVEPOINT on an already-poisoned # session would itself raise and MASK that original exception — the # precise outcome the comment above forbids. Named locks are # session-scoped and unaffected by savepoints anyway, and session-end # auto-release is the backstop. # savepoint-exempt: savepoint would mask the propagating exception try: await db.execute( text("SELECT RELEASE_LOCK(:name)"), {"name": lock_name} ) except Exception: logger.exception( "Failed to RELEASE_LOCK for %s in close_month finally — " "session-end auto-release will handle. Original " "exception (if any) preserved.", lock_name, ) async def reopen_month( db: AsyncSession, period: str, *, actor_id: uuid.UUID, ) -> dict[str, Any]: """Reopen a closed period so it can be re-closed with corrected data. Snapshot-then-delete: every row of ``software_monthly_bills`` for ``period`` is copied into ``software_monthly_bill_history`` BEFORE the live rows are removed. Each snapshot batch carries the same ``supersedes_seq`` (1 for the first reopen of a period, 2 for the second, …) so finance can reconstruct any prior version of the bill by querying ``WHERE period=? AND supersedes_seq=?``. No-op (idempotent) when the period has no live rows — returns ``snapshot_count=0`` and a fresh ``supersedes_seq`` is NOT consumed (the seq advances only when there's something to snapshot). Wraps snapshot-insert + delete in a single flush ordering: write snapshots → flush → delete live rows → flush. The caller commits. Order matters because both tables are touched in one transaction; a flush failure on the snapshot side aborts before any DELETE. Concurrency: shares the SAME advisory lock as ``close_month`` (``helix.billing.close.{period}``) — reopen and close on the same period are mutually exclusive. Without this, two parallel reopens (admin double-click on the "重新打开" button, or admin + auto-job concurrency) could both read the same MAX(supersedes_seq), both compute next_seq=N+1, and both INSERT history rows with the same composite key. Migration 119 also adds a UNIQUE INDEX on ``(period, supersedes_seq, original_bill_id)`` as belt-and-braces. """ _parse_period(period) lock_name = f"helix.billing.close.{period}" # Same 5s acquisition window as ``close_month`` — see comment there # for why fail-fast (0s) is worse than a brief wait for benign # admin-vs-cron overlap. lock_acquired = ( await db.execute( text("SELECT GET_LOCK(:name, 5)"), {"name": lock_name}, ) ).scalar() if lock_acquired is None: # GET_LOCK returns NULL on lock-server / OS error (e.g. lock # subsystem crashed mid-call). Distinguish from a busy lock # so ops sees the real failure mode in the response. Round-3 # logic-scan P1. raise AppException( code="PERIOD_LOCK_INFRASTRUCTURE_ERROR", message=( f"Failed to acquire reopen lock for {period}: MySQL " "advisory-lock subsystem returned NULL. Retry; if " "persistent, escalate." ), status_code=503, ) if lock_acquired != 1: raise AppException( code="PERIOD_REOPEN_IN_PROGRESS", message=( f"Period {period} is currently being closed or reopened " "in another session. Try again shortly." ), status_code=409, ) # try/finally so a mid-flight raise (snapshot insert FK violation, # delete failure, etc.) still releases the advisory lock — without # this the lock would only auto-release on session close, which # for connection-pool reuse / async retries can hold past the # failed reopen and block the next admin attempt. Round-2 audit # P1. try: return await _reopen_month_locked( db, period, actor_id=actor_id, ) finally: # Inner try/except — see ``close_month``'s finally for why. A # RELEASE_LOCK failure here must NOT mask the original # exception from ``_reopen_month_locked``. # savepoint-exempt: same as close_month's finally — a SAVEPOINT on an # already-poisoned session would mask the original exception. try: await db.execute( text("SELECT RELEASE_LOCK(:name)"), {"name": lock_name}, ) except Exception: logger.exception( "Failed to RELEASE_LOCK for %s in reopen_month finally — " "session-end auto-release will handle. Original " "exception (if any) preserved.", lock_name, ) async def _reopen_month_locked( db: AsyncSession, period: str, *, actor_id: uuid.UUID, ) -> dict[str, Any]: """Inner body of ``reopen_month`` once the advisory lock is held. Split out so the lock's RELEASE in ``reopen_month``'s ``finally`` fires on every code path (early-return, raise, success). """ # 1. Pull live rows that are about to be cleared. live_rows = ( await db.execute( select(SoftwareMonthlyBill).where(SoftwareMonthlyBill.period == period) ) ).scalars().all() if not live_rows: # Idempotent no-op — nothing to snapshot, nothing to delete. # Don't burn a supersedes_seq when there's no batch. return { "period": period, "deleted": 0, "snapshot_count": 0, "snapshot_total": Decimal("0.00"), "supersedes_seq": None, } # 2. Determine the next supersedes_seq for this period. prior_max = ( await db.execute( select(func.max(SoftwareMonthlyBillHistory.supersedes_seq)).where( SoftwareMonthlyBillHistory.period == period, ) ) ).scalar() next_seq = (prior_max or 0) + 1 # 3. Snapshot each live row into history. ``superseded_at`` is a # single ``now()`` for the whole batch so all rows from one reopen # share the same timestamp — convenient for "show me batch N". superseded_at = datetime.now(UTC) snapshot_total = Decimal("0.00") for r in live_rows: # Migration 116 backfills every existing row to 'CNY' on # add-column; direct access is safe post-migration. row_currency = (r.currency or "CNY").upper() db.add( SoftwareMonthlyBillHistory( id=uuid.uuid4(), period=r.period, supersedes_seq=next_seq, original_bill_id=r.id, line_type=getattr(r, "line_type", "seat") or "seat", allocation_id=r.allocation_id, software_id=r.software_id, plan_id=r.plan_id, charge_id=r.charge_id, quantity=r.quantity, user_id=r.user_id, software_name=r.software_name, plan_name=r.plan_name, user_name=r.user_name, department=r.department, reason=r.reason, price=r.price, billing_cycle=r.billing_cycle, currency=row_currency, amount=r.amount, allocated_at=r.allocated_at, revoked_at=r.revoked_at, original_finalized_at=r.finalized_at, original_finalized_by=r.finalized_by, superseded_at=superseded_at, superseded_by=actor_id, ) ) snapshot_total += r.amount # Force snapshots to land before the DELETE — if the INSERT fails # (constraint, FK, etc.) the transaction aborts before the live # rows are removed and the period stays intact. await db.flush() # 4. DELETE the live rows. deleted = ( await db.execute( delete(SoftwareMonthlyBill).where(SoftwareMonthlyBill.period == period) ) ).rowcount or 0 await db.flush() return { "period": period, "deleted": deleted, "snapshot_count": len(live_rows), "snapshot_total": _quantize(snapshot_total), "supersedes_seq": next_seq, } async def _resolve_department_user_ids( db: AsyncSession, department: str, ) -> set[uuid.UUID]: """Resolve a department name to every user whose primary org membership is that department or any descendant. Mirrors the subtree semantics of ``/cost-stats/department``: asking for "Business Operations" must pick up everyone in "IT Services & Security" and every other descendant. Delegates to the shared ``app.core.org_scope`` primitive (single source of truth) instead of re-implementing the breadcrumb match — a hand-rolled copy here had drifted and over-matched prefix-similar siblings ("Security" pulling in "Security & Compliance"). """ from app.core.org_scope import department_membership_subtree_clause from app.identity.person_models import IdentityOrgMembership rows = ( await db.execute( select(IdentityOrgMembership.person_id) .where(IdentityOrgMembership.is_primary == True) # noqa: E712 .where(IdentityOrgMembership.deleted_at.is_(None)) .where(department_membership_subtree_clause(department)) ) ).all() return {r.person_id for r in rows} async def list_period_bills( db: AsyncSession, period: str, *, user_ids: list[uuid.UUID] | None = None, software_id: uuid.UUID | None = None, department: str | None = None, ) -> dict[str, Any]: """Fetch the finalised bill for a period. If the period is not yet closed, falls back to the live preview (with ``status=draft``). Department filter is **subtree-expanded** to match the ``/cost-stats/department`` semantics — asking for "Business Operations" pulls in every descendant dept. Expansion resolves against the current org tree (not the frozen dept snapshot), since the dept tree rarely changes and admins expect "today's Business Operations" when they click the filter.""" _parse_period(period) # Expand department → user_ids if filtering by department. Intersect # with any already-scoped user_ids the caller passed. Empty set means # "no users in scope" — short-circuit so we don't issue a pointless # ``IN ()`` query. effective_user_ids: list[uuid.UUID] | None = user_ids if department: dept_user_ids = await _resolve_department_user_ids(db, department) if user_ids is not None: dept_user_ids &= set(user_ids) effective_user_ids = list(dept_user_ids) if not effective_user_ids: return { "period": period, "status": "draft", "total_amount": Decimal("0.00"), "by_currency": {}, "total_currency_count": 0, "currency_warning": False, "active_seats": 0, "finalized_at": None, "finalized_by": None, "items": [], } # The vendor-authoritative seat count (``vendor_license_holders`` from # the seat-sync table) is an ORG-WIDE number — it mirrors the vendor's # admin console total for the whole company. It is only meaningful, and # only safe to expose, on the whole-company admin view (``/all``, which # passes no ``user_ids``/``department``). On a scoped view (``/my``, # ``/subordinates``, ``/department``) returning it would (a) leak the # company-wide paid-seat count of every product the caller holds to a # self-scope endpoint, and (b) report a wildly wrong "生效坐席" (e.g. an # employee viewing their own bill would see the org-wide seat total # instead of 1). Scoped views fall back to the distinct-user count # within scope and do NOT surface ``vendor_seats_by_software``. scope_is_global = effective_user_ids is None rows = ( await db.execute( select(SoftwareMonthlyBill) .where(SoftwareMonthlyBill.period == period) .where( *( [SoftwareMonthlyBill.user_id.in_(effective_user_ids)] if effective_user_ids is not None else [] ) ) .where( *( [SoftwareMonthlyBill.software_id == software_id] if software_id else [] ) ) .order_by(SoftwareMonthlyBill.software_name.asc(), SoftwareMonthlyBill.user_name.asc()) ) ).scalars().all() if not rows: # Not closed yet — synthesise a draft on the fly. Apply the # same filters client-side since preview doesn't accept them. draft = await preview_current_or_draft(db, period) items = draft["items"] if effective_user_ids is not None: allowed = set(effective_user_ids) items = [it for it in items if it["user_id"] in allowed] if software_id: items = [it for it in items if it.get("software_id") == software_id] by_currency, single_total, warn = _aggregate_by_currency(items) # Vendor-authoritative seat count (matches the drawer's # "厂商许可"). Recompute the family-deduped map narrowed to the # still-present software_ids after filtering so the top-level # count drops to 0 when filters carve everything out. See # ``preview_current_or_draft`` for the full rationale. sw_present = {it["software_id"] for it in items if it.get("software_id")} vendor_seats = ( await _vendor_seats_by_software(db, sw_present) if scope_is_global else {} ) active_seats = _seat_count_from_lines(items, vendor_seats) return { "period": period, "status": "draft", "total_amount": single_total, "by_currency": {k: str(v) for k, v in by_currency.items()}, "total_currency_count": len(by_currency), "currency_warning": warn, "active_seats": active_seats, "vendor_seats_by_software": _flatten_vendor_seats(vendor_seats), "finalized_at": None, "finalized_by": None, "items": items, } # Finalised: build response from stored rows. Status per row comes # from revoked_at (historical snapshot). # # ``user_email`` is NOT snapshotted in ``software_monthly_bill`` — # the bill row freezes name/department for historical fidelity, but # email is looked up live so admins can search the closed-period # table by current email (e.g. typing ``kevin.lu@pingcap.com`` into # the keyword box must match the user even on a 6-month-old bill). # If the person row has been hard-deleted, the email field falls # back to ``None`` and the row still renders with name/department. bill_user_ids = {r.user_id for r in rows if r.user_id is not None} user_email_by_id: dict[uuid.UUID, str | None] = {} if bill_user_ids: # Chunk the IN clause defensively — TiDB recommends keeping IN # under ~1000 elements for a clean IndexLookup plan, and a # months-old period with thousands of seats could otherwise # land here with a 5k+ element IN. The per-chunk query is # keyed on the PK index (``IdentityPerson.id``), so each chunk # is one cheap point-lookup batch. _CHUNK = 1000 candidate_ids = list(bill_user_ids) for start in range(0, len(candidate_ids), _CHUNK): chunk = candidate_ids[start : start + _CHUNK] person_rows = ( await db.execute( select(IdentityPerson.id, IdentityPerson.work_email).where( IdentityPerson.id.in_(chunk), ) ) ).all() for pid, email in person_rows: user_email_by_id[pid] = email # Resolve each row's BU (top-level org node) by current primary # membership — same derivation the ``bu`` filter uses, so a closed # bill rolls up to today's BU. The frozen ``department`` leaf is # historical fidelity; BU is intentionally current (org reshuffles # are rare and admins expect "today's BU" the same way the dept # filter expands against today's tree). from app.identity.bu_helpers import bu_for_user_ids bu_by_user = await bu_for_user_ids(db, bill_user_ids) if bill_user_ids else {} items: list[dict[str, Any]] = [] period_end = _parse_period(period)[1] end_dt = datetime.combine(period_end, datetime.min.time(), tzinfo=UTC) # "生效坐席" — see ``preview_current_or_draft`` for the rationale. # Closed periods read from the frozen ``software_monthly_bill`` snapshot # rather than recomputing seat allocations, but the workbench shows the # SAME column whether the period is closed or live, so the count must # use the same definition. Vendor seat-sync data is point-in-time (it # mirrors today's vendor state, not the closed period's), but on a # closed period the on-staff fleet has not changed enough between # close-time and now to invalidate the cell — the drawer drill-down # uses the same snapshot, so they continue to agree. sw_present: set[uuid.UUID] = {r.software_id for r in rows if r.software_id} vendor_seats = ( await _vendor_seats_by_software(db, sw_present) if scope_is_global else {} ) active_seats = _seat_count_from_lines( [ # ``line_type`` MUST be carried through — _seat_count_from_lines # skips ``addon`` rows (one-time add-on charges are not seats). # Omitting it here counted every add-on charge as a seat, so the # 生效坐席 headline silently jumped by the add-on-row count the # moment a period closed (violating "seat count == vendor truth"). { "software_id": r.software_id, "user_id": r.user_id, "line_type": r.line_type, } for r in rows ], vendor_seats, ) for r in rows: # Same TZ-naive coercion as ``_build_lines_for_period`` — MySQL # DATETIME columns lose timezone info on read even when the ORM # declares ``DateTime(timezone=True)``. Without normalising, # a closed-period reload after the first migration-097 cron run # 500'd here too. See companion fix at line ~189. rev = r.revoked_at if rev is not None and rev.tzinfo is None: rev = rev.replace(tzinfo=UTC) revoked = rev is not None and rev < end_dt line_status = "revoked" if revoked else "active" items.append({ "id": r.id, "period": r.period, "allocation_id": r.allocation_id, "software_id": r.software_id, "software_name": r.software_name, "plan_id": r.plan_id, "plan_name": r.plan_name, "user_id": r.user_id, "user_name": r.user_name, "user_email": user_email_by_id.get(r.user_id), "department": r.department, "bu": bu_by_user.get(str(r.user_id)), "reason": r.reason, "price": r.price, "billing_cycle": r.billing_cycle, "currency": (r.currency or "CNY").upper(), "amount": r.amount, "allocated_at": r.allocated_at, "revoked_at": r.revoked_at, "status": line_status, "finalized_at": r.finalized_at, }) finalized_at = rows[0].finalized_at if rows else None finalized_by = rows[0].finalized_by if rows else None by_currency, single_total, warn = _aggregate_by_currency(items) return { "period": period, "status": "finalized", "total_amount": single_total, "by_currency": {k: str(v) for k, v in by_currency.items()}, "total_currency_count": len(by_currency), "currency_warning": warn, "active_seats": active_seats, "vendor_seats_by_software": _flatten_vendor_seats(vendor_seats), "finalized_at": finalized_at, "finalized_by": finalized_by, "items": items, } async def list_known_periods(db: AsyncSession, *, months: int = 12) -> list[dict[str, Any]]: """Recent periods with close status. Returns at most ``months`` entries including current + prior months. Used to populate the month picker in the UI so the admin knows which months are closed.""" today = datetime.now(UTC).date() periods: list[str] = [] y, m = today.year, today.month for _ in range(months): periods.append(f"{y:04d}-{m:02d}") m -= 1 if m == 0: m = 12 y -= 1 # Pull totals for any period that IS closed in one shot. Live # previews are computed only on demand (per-period open request) — # doing 12× preview synthesis at list time would be silly. # Group by ``(period, currency)`` so a period with mixed currencies # surfaces a per-currency breakdown — the single-decimal # ``total_amount`` collapses to ``None`` in that case. agg = ( await db.execute( select( SoftwareMonthlyBill.period, SoftwareMonthlyBill.currency, func.count(SoftwareMonthlyBill.id).label("line_count"), func.sum(SoftwareMonthlyBill.amount).label("total_amount"), func.max(SoftwareMonthlyBill.finalized_at).label("finalized_at"), func.max(SoftwareMonthlyBill.finalized_by).label("finalized_by"), ) .where(SoftwareMonthlyBill.period.in_(periods)) .group_by(SoftwareMonthlyBill.period, SoftwareMonthlyBill.currency) ) ).all() # Per-period distinct (user, software) pair count — the historical # equivalent of the live view's vendor-seat-aligned "生效坐席". We # can't read ``software_seat_sync_stat`` for a closed period (it only # ever holds CURRENT vendor state, not a per-close snapshot), so use # distinct (user, software) pairs from the frozen bill — one person # held one software once = one seat, exactly the "people billed this # month" reading the picker tooltip wants. Cheap enough to run as a # second aggregate over the same row set since the row count for # 12 months stays bounded. seat_agg = ( await db.execute( select( SoftwareMonthlyBill.period, func.count( func.distinct( func.concat( func.coalesce(SoftwareMonthlyBill.software_id, ""), ":", SoftwareMonthlyBill.user_id, ) ) ).label("seat_count"), ) # Exclude add-on charge rows — they are not seats (mirror # _seat_count_from_lines). Without this the month-picker seat # tooltip over-counts by the add-on-row count for any month # whose software accrued one-time add-on charges. .where( SoftwareMonthlyBill.period.in_(periods), SoftwareMonthlyBill.line_type != "addon", ) .group_by(SoftwareMonthlyBill.period) ) ).all() seat_count_by_period: dict[str, int] = { row.period: int(row.seat_count or 0) for row in seat_agg } # Fold per-(period, currency) rows back into per-period buckets. by_period: dict[str, dict[str, Any]] = {} for row in agg: bucket = by_period.setdefault(row.period, { "by_currency": {}, "line_count": 0, "finalized_at": None, "finalized_by": None, }) cur = (row.currency or "CNY").upper() bucket["by_currency"][cur] = ( bucket["by_currency"].get(cur, Decimal("0.00")) + (row.total_amount or Decimal("0.00")) ) bucket["line_count"] += int(row.line_count or 0) # Both finalized_at / finalized_by are uniform per period (one # close = one timestamp); pick whichever non-null value we see. if row.finalized_at is not None and bucket["finalized_at"] is None: bucket["finalized_at"] = row.finalized_at if row.finalized_by is not None and bucket["finalized_by"] is None: bucket["finalized_by"] = row.finalized_by out: list[dict[str, Any]] = [] current = _current_period() for p in periods: bucket = by_period.get(p) if bucket: cur_totals = {k: _quantize(v) for k, v in bucket["by_currency"].items()} single_total: Decimal | None if len(cur_totals) <= 1: single_total = next(iter(cur_totals.values()), Decimal("0.00")) else: single_total = None out.append({ "period": p, "status": "finalized", "total_amount": single_total, "by_currency": {k: str(v) for k, v in cur_totals.items()}, "total_currency_count": len(cur_totals), "currency_warning": len(cur_totals) > 1, # Distinct (user, software) pairs — matches the per-period # detail endpoints' "people billed this month" definition. # 2026-05-30 follow-up to PR #579: the previous ``line_count`` # interpretation inflated to bill-line count (175 for JSM's # 72 actual seats), which contradicted the drill-down drawer. "active_seats": seat_count_by_period.get(p, 0), "finalized_at": bucket["finalized_at"], "finalized_by": bucket["finalized_by"], }) else: out.append({ "period": p, "status": "draft" if p >= current else "available", "total_amount": None, "by_currency": {}, "total_currency_count": 0, "currency_warning": False, "active_seats": None, "finalized_at": None, "finalized_by": None, }) return out # --------------------------------------------------------------------------- # Range summary — "这些软件在这段时间分别一共花了多少" # --------------------------------------------------------------------------- # The per-period views above answer "what did month M cost". Finance's other # standing question is the cross-month one: pick a month range + (optionally) # a few software, get each software's total for the whole range. Before this # the only way was opening 12 monthly tabs and adding the numbers by hand. # # Caliber is deliberately IDENTICAL to the single-month view so the two # surfaces can never disagree: # * a closed month reads the FROZEN ``software_monthly_bills`` snapshot, so # a later price edit / rename never moves a month that's already closed; # * a month with no frozen rows is computed live — the same preview the # monthly tab labels "draft". In practice that's only the current month, # because the ``software_billing_auto_close`` cron freezes last month on # the 1st; # * amounts stay in their NATIVE currency, bucketed per currency — never # FX-converted, never summed across currencies (a CNY plan and a USD plan # are not addable). The FX-normalised USD lens is a different surface # (``/finance/spend/*``), and it deliberately excludes ``category='ai'`` # software; this one does not. # * add-on charge lines count toward money (they are real spend) but not # toward the billed-people count — mirrors ``_seat_count_from_lines``. # Hard ceiling on the month span. 24 months covers "上一个财年 + 今年至今", # which is the widest question finance has actually asked; beyond that the # frozen-row scan stops being a bounded read. RANGE_MAX_MONTHS = 24 # A live preview is a full allocation scan for that month, so a range whose # months were never closed could fan out into 24 scans. Cap the live work — # and REPORT the months we refused to compute (never silently drop them), so # the reader sees "close these months first" instead of trusting a short # total. Practically unreachable while the auto-close cron is healthy. RANGE_MAX_LIVE_PERIODS = 3 # Concurrency gate for live-preview scans across ALL in-flight # range-summary requests (#2191). A live preview is a full seat # allocation scan; ``RANGE_MAX_LIVE_PERIODS`` bounds how many ONE # request may trigger, but nothing bounded how many requests could do it # at once — and ``view=my`` is open to every employee. Process-local (one # per worker), which is the right granularity: the thing being protected # is this process's DB connections and CPU. _LIVE_PREVIEW_CONCURRENCY = 2 _LIVE_PREVIEW_SEMAPHORE = asyncio.Semaphore(_LIVE_PREVIEW_CONCURRENCY) def _enumerate_periods(start: str, end: str) -> list[str]: """Inclusive ascending ``YYYY-MM`` list between the two endpoints. Accepts them in either order (``YYYY-MM`` strings sort chronologically). Raises ``RANGE_TOO_WIDE`` past :data:`RANGE_MAX_MONTHS` months. """ _parse_period(start) _parse_period(end) lo, hi = sorted([start, end]) out: list[str] = [] year, month = int(lo[:4]), int(lo[5:]) for _ in range(RANGE_MAX_MONTHS): period = f"{year:04d}-{month:02d}" out.append(period) if period == hi: return out month += 1 if month == 13: month, year = 1, year + 1 raise AppException( code="RANGE_TOO_WIDE", message=( f"Range spans more than {RANGE_MAX_MONTHS} months " f"({lo} → {hi}); narrow the range." ), status_code=400, ) def _range_bucket( acc: dict[str, dict[str, Any]], software_id: Any, software_name: str | None, ) -> dict[str, Any]: """Get-or-create the per-software accumulator bucket. Keyed on ``software_id`` (never the name) so a software renamed mid-range stays ONE row; the display label is taken from the latest period that carried a name (see ``summarize_range``). """ key = str(software_id) if software_id else "" bucket = acc.get(key) if bucket is None: bucket = { "software_id": str(software_id) if software_id else None, "software_name": software_name, "by_currency": {}, # {currency: Decimal} range total "months": {}, # {period: {currency: Decimal}} "users": set(), # distinct billed user_ids (non-addon) } acc[key] = bucket return bucket def _range_add( bucket: dict[str, Any], period: str, currency: str | None, amount: Any, ) -> None: cur = (currency or "CNY").upper() amt = amount if isinstance(amount, Decimal) else Decimal(str(amount or 0)) bucket["by_currency"][cur] = bucket["by_currency"].get(cur, Decimal("0.00")) + amt month = bucket["months"].setdefault(period, {}) month[cur] = month.get(cur, Decimal("0.00")) + amt async def _detect_currency_drift( db: AsyncSession, acc: dict[str, dict[str, Any]], frozen_periods: set[str], ) -> list[dict[str, Any]]: """Frozen months whose currency label disagrees with the software's CURRENT plan currency. Deliberately conservative — a hit must be unambiguous, or the banner becomes noise the reader learns to skip: * only FROZEN months are examined (a live month is computed FROM the plan, so it can never disagree with it); * only software whose live plans all share ONE currency; a product that genuinely bills two currencies has no single "right" label to compare against. Reports; never rewrites. Correcting a frozen financial snapshot is a deliberate, audited data fix (``scripts/fix_ai_seat_bill_currency_2026_07_30.py``), not something a read endpoint should do behind the reader's back. """ software_ids = [ uuid.UUID(bucket["software_id"]) for bucket in acc.values() if bucket["software_id"] ] if not software_ids or not frozen_periods: return [] plan_rows = ( await db.execute( select( SoftwareLicensePlan.software_id, SoftwareLicensePlan.currency, ) .where( SoftwareLicensePlan.software_id.in_(software_ids), SoftwareLicensePlan.status == "active", ) .distinct() ) ).all() live_currencies: dict[str, set[str]] = {} for sw_id, currency in plan_rows: live_currencies.setdefault(str(sw_id), set()).add( (currency or "CNY").upper() ) drift: list[dict[str, Any]] = [] for key, bucket in acc.items(): live = live_currencies.get(key) if not live or len(live) != 1: continue plan_currency = next(iter(live)) mismatched = sorted( { (period, currency) for period, per_cur in bucket["months"].items() if period in frozen_periods for currency in per_cur if currency != plan_currency } ) if mismatched: drift.append({ "software_id": bucket["software_id"], "software_name": bucket["software_name"], "plan_currency": plan_currency, "periods": [ {"period": p, "currency": c} for p, c in mismatched ], }) drift.sort(key=lambda d: d["software_name"] or "") return drift async def summarize_range( db: AsyncSession, start: str, end: str, *, user_ids: list[uuid.UUID] | None = None, software_ids: list[uuid.UUID] | None = None, department: str | None = None, ) -> dict[str, Any]: """Per-software spend totals across an inclusive ``YYYY-MM`` range. See the section comment above for the caliber contract. ``user_ids`` narrows to a scope (``None`` = whole company), ``software_ids`` narrows to specific catalog rows, ``department`` is subtree-expanded exactly like :func:`list_period_bills`. """ periods = _enumerate_periods(start, end) # Department → user_ids, intersected with any caller-supplied scope. # Mirrors list_period_bills, including the empty short-circuit. effective_user_ids: list[uuid.UUID] | None = user_ids if department: dept_user_ids = await _resolve_department_user_ids(db, department) if user_ids is not None: dept_user_ids &= set(user_ids) effective_user_ids = list(dept_user_ids) sw_filter = [sid for sid in (software_ids or []) if sid] # Which months are frozen? Deliberately UNFILTERED by scope/software — # "this month has no rows at all" (never closed) is a different fact # from "this month is closed but the filter matched nothing", and # conflating them would silently live-preview a closed month. frozen_periods = set( ( await db.execute( select(SoftwareMonthlyBill.period) .where(SoftwareMonthlyBill.period.in_(periods)) .distinct() ) ).scalars().all() ) # A month that hasn't happened yet is NEVER computed. ``preview_current_ # or_draft`` happily bills every currently-active seat for any period it # is handed, so live-previewing 2026-10 in July would fabricate three # months of cost, label it "实时", AND burn the live-preview budget that # the real current month needs. Future months are reported as such and # contribute nothing. current = _current_period() future_periods = [p for p in periods if p > current] # Months with no frozen rows → live preview, newest first (the current # month is the one that matters). Anything past the cap is reported, # not silently dropped. missing = [ p for p in periods if p not in frozen_periods and p not in future_periods ] live_periods = sorted(missing)[-RANGE_MAX_LIVE_PERIODS:] excluded_periods = [p for p in missing if p not in live_periods] if excluded_periods: logger.warning( "software_billing.range_summary: %d unclosed month(s) excluded " "(live-preview cap %d): %s", len(excluded_periods), RANGE_MAX_LIVE_PERIODS, excluded_periods, ) period_status = { p: ( "finalized" if p in frozen_periods else "future" if p in future_periods else "live" if p in live_periods else "excluded" ) for p in periods } envelope: dict[str, Any] = { "start": periods[0], "end": periods[-1], "periods": periods, "period_status": period_status, "live_periods": live_periods, "excluded_periods": excluded_periods, # Months later than the current one. Not computed, not in the totals # — a range that reaches into the future is a picker slip, and a # fabricated projection is the last thing a cost total should carry. "future_periods": future_periods, } if effective_user_ids is not None and not effective_user_ids: # No people in scope — nothing can have billed. Still report the # month statuses so the UI renders its close-status legend rather # than an unexplained blank. return { **envelope, "software": [], "totals": {"by_currency": {}, "months": {}}, "currency_warning": False, "currency_drift": [], "distinct_users": 0, } scope_clauses = [] if effective_user_ids is not None: scope_clauses.append(SoftwareMonthlyBill.user_id.in_(effective_user_ids)) if sw_filter: scope_clauses.append(SoftwareMonthlyBill.software_id.in_(sw_filter)) acc: dict[str, dict[str, Any]] = {} # Latest period that named each software, so a mid-range rename shows # the CURRENT name on the (single) row rather than an arbitrary one. name_period: dict[str, str] = {} if frozen_periods: frozen_list = sorted(frozen_periods) money_rows = ( await db.execute( select( SoftwareMonthlyBill.period, SoftwareMonthlyBill.software_id, SoftwareMonthlyBill.software_name, SoftwareMonthlyBill.currency, func.sum(SoftwareMonthlyBill.amount).label("amount"), ) .where(SoftwareMonthlyBill.period.in_(frozen_list), *scope_clauses) .group_by( SoftwareMonthlyBill.period, SoftwareMonthlyBill.software_id, SoftwareMonthlyBill.software_name, SoftwareMonthlyBill.currency, ) ) ).all() for row in money_rows: m = row._mapping key = str(m["software_id"]) if m["software_id"] else "" bucket = _range_bucket(acc, m["software_id"], m["software_name"]) if m["software_name"] and m["period"] >= name_period.get(key, ""): bucket["software_name"] = m["software_name"] name_period[key] = m["period"] _range_add(bucket, m["period"], m["currency"], m["amount"]) # Distinct (software, person) pairs — the range's "计费人数". Bounded # by people × software regardless of how many months are in range, # so this stays a small result set even at 24 months. Fetched as # PAIRS (not a COUNT) because the live-previewed months below must # union into the same de-duplicated set — a count plus a set can't # be merged honestly. pair_rows = ( await db.execute( select( SoftwareMonthlyBill.software_id, SoftwareMonthlyBill.user_id, ) .where( SoftwareMonthlyBill.period.in_(frozen_list), SoftwareMonthlyBill.line_type != "addon", *scope_clauses, ) .distinct() ) ).all() for sw_id, uid in pair_rows: _range_bucket(acc, sw_id, None)["users"].add(str(uid)) allowed_users = set(effective_user_ids) if effective_user_ids is not None else None allowed_software = {str(s) for s in sw_filter} for period in live_periods: # Bounded concurrency around the live preview (#2191). Each call # is a FULL seat-allocation scan, this endpoint runs up to # ``RANGE_MAX_LIVE_PERIODS`` of them per request, and # ``view=my`` carries no role gate — so if the auto-close cron # breaks and several months sit unclosed, any employee # rapid-clicking the 区间汇总 tab multiplies the scan load with # nothing in the way (the single-month endpoint at least does # one per request). The cap bounds the CONCURRENT scans across # all in-flight requests; queued callers wait rather than pile # on. Not a cache — a cache here would need the same # invalidation discipline the frozen-vs-live split already # encodes, and the issue's own read is that this is a load # guard, not a latency problem. async with _LIVE_PREVIEW_SEMAPHORE: draft = await preview_current_or_draft(db, period) for line in draft["items"]: if allowed_users is not None and line["user_id"] not in allowed_users: continue sw_id = line.get("software_id") if allowed_software and str(sw_id) not in allowed_software: continue key = str(sw_id) if sw_id else "" bucket = _range_bucket(acc, sw_id, line.get("software_name")) if line.get("software_name") and period >= name_period.get(key, ""): bucket["software_name"] = line["software_name"] name_period[key] = period _range_add(bucket, period, line.get("currency"), line.get("amount")) if line.get("line_type") != "addon": bucket["users"].add(str(line["user_id"])) # ── Currency-label drift ── # A frozen month keeps the currency its plan carried at close time — by # design. But when that label disagrees with what the software charges # TODAY, the same seat at the same price shows up under two currencies # across the range and the reader has no way to tell "we really switched # billing currency" from "the old label is wrong". That happened for real # (migration 117 blanket-backfilled every plan + bill to CNY on # 2026-05-05, mislabelling the USD-priced Claude / OPEN AI seats until # the 2026-06 close), and nothing surfaced it — a human spotted ¥ next to # $ on this very table. So name it here instead of waiting for eyes. drift = await _detect_currency_drift(db, acc, frozen_periods) # ── Shape the response ── totals_by_currency: dict[str, Decimal] = {} totals_by_month: dict[str, dict[str, Decimal]] = {} all_users: set[str] = set() # (sort_weight, row) pairs — see the sort note below. weighted_rows: list[tuple[Decimal, str, dict[str, Any]]] = [] for bucket in acc.values(): for cur, amt in bucket["by_currency"].items(): totals_by_currency[cur] = totals_by_currency.get(cur, Decimal("0.00")) + amt for period, per_cur in bucket["months"].items(): month_bucket = totals_by_month.setdefault(period, {}) for cur, amt in per_cur.items(): month_bucket[cur] = month_bucket.get(cur, Decimal("0.00")) + amt all_users |= bucket["users"] weighted_rows.append(( # Biggest spender first. This ordering weight sums across # currencies, which is NOT a money figure and is never rendered # as one — it only decides row order (the frontend's table sorts # on the same basis, for the same reason). sum(bucket["by_currency"].values(), Decimal("0.00")), bucket["software_name"] or "", { "software_id": bucket["software_id"], "software_name": bucket["software_name"], "by_currency": { k: str(_quantize(v)) for k, v in sorted(bucket["by_currency"].items()) }, "months": { p: {k: str(_quantize(v)) for k, v in sorted(per_cur.items())} for p, per_cur in sorted(bucket["months"].items()) }, "distinct_users": len(bucket["users"]), }, )) weighted_rows.sort(key=lambda pair: (-pair[0], pair[1])) software_rows = [row for _, _, row in weighted_rows] return { **envelope, "software": software_rows, "totals": { "by_currency": { k: str(_quantize(v)) for k, v in sorted(totals_by_currency.items()) }, "months": { p: {k: str(_quantize(v)) for k, v in sorted(per_cur.items())} for p, per_cur in sorted(totals_by_month.items()) }, }, "currency_warning": len(totals_by_currency) > 1, # Frozen months whose currency label disagrees with the software's # current plan currency — see ``_detect_currency_drift``. "currency_drift": drift, "distinct_users": len(all_users), } __all__ = [ "RANGE_MAX_LIVE_PERIODS", "RANGE_MAX_MONTHS", "close_month", "get_period_status", "list_known_periods", "list_period_bills", "period_overlap_clauses", "preview_current_or_draft", "reopen_month", "summarize_range", ]