"""Field-level masking helpers for sensitive identifiers.

Centralised so every surface (HTTP responses, MCP tools, exports) masks
the *same* way. Always mask in the BACKEND — shipping plaintext to the
browser and masking in the UI leaves the real value one devtools call
away, so the masking has to happen before the value crosses the wire.

These are deterministic, allocation-light, and ``None``-safe: a falsy
input is returned unchanged so callers can pipe optional columns through
without a guard.
"""
from __future__ import annotations

import re
from typing import Any

# Known E.164 calling-code prefixes, used by ``mask_phone`` to split a
# delimiter-less ``+CC…`` number on a real country-code boundary. Covers
# the 1-2-digit codes that the old greedy 3-digit grab mangled, plus the
# 3-digit codes relevant to PingCAP's footprint. Longest-match wins.
_KNOWN_COUNTRY_CODES: frozenset[str] = frozenset({
    "1", "7", "20", "27", "30", "31", "32", "33", "34", "39", "44",
    "49", "52", "55", "60", "61", "62", "63", "64", "65", "66", "81",
    "82", "84", "86", "90", "91", "92", "234", "351", "852", "853",
    "886", "966", "971",
})


def mask_phone(raw: str | None) -> str | None:
    """Mask a phone number, preserving an optional ``+`` country code and
    the last 4 digits.

    ``+8613812345678`` → ``+86 138****5678``
    ``13812345678``    → ``138****5678``
    ``123``            → ``***`` (too short to split meaningfully)
    """
    if not raw:
        return raw
    s = raw.strip()
    cc = ""
    rest = s
    if s.startswith("+"):
        # An E.164 number carries no delimiter (``+8613812345678``), so a
        # greedy "take N digits" split mis-buckets the country code — it
        # would grab ``+861`` for a +86 number and shift the first national
        # digit into the visible head. Resolve against a known-prefix table
        # instead, longest-match first, so the split lands on a real CC
        # boundary. Unknown prefixes fall back to a 2-digit code (the common
        # case) rather than greedily eating 3.
        digit_run = "".join(c for c in s[1:] if c.isdigit())
        for k in sorted(_KNOWN_COUNTRY_CODES, key=len, reverse=True):
            if digit_run.startswith(k):
                cc, rest = "+" + k, digit_run[len(k):]
                break
        else:
            cc, rest = "+" + digit_run[:2], digit_run[2:]
    digits = "".join(c for c in rest if c.isdigit())
    # head(3)+tail(4)=7: at 7 or fewer national digits the head and tail
    # slices OVERLAP, so ``head + "****" + tail`` would echo every original
    # digit back while looking masked (e.g. ``1234567`` → ``123****4567``).
    # Fully mask anything too short to hide at least one digit, mirroring
    # the ``len <= head+tail`` guard the other two maskers already have.
    if len(digits) <= 7:
        masked = "*" * len(digits)
        return f"{cc} {masked}".strip() if cc else masked
    head, tail = digits[:3], digits[-4:]
    middle = "*" * max(4, len(digits) - len(head) - len(tail))
    masked = f"{head}{middle}{tail}"
    return f"{cc} {masked}".strip() if cc else masked


def mask_tail(raw: str | None, *, visible: int = 4) -> str | None:
    """Mask everything but the last ``visible`` characters.

    Used for bank accounts and tax identifiers, where the trailing digits
    are enough to recognise a value you already know without disclosing it.

    ``6225880112345678`` → ``************5678``
    ``1234``             → ``****`` (≤ visible ⇒ fully masked)
    """
    if not raw:
        return raw
    s = raw.strip()
    # ``visible <= 0`` must fully mask. Without this guard ``s[-0:]`` is the
    # WHOLE string, so ``mask_tail(secret, visible=0)`` — the natural way to
    # ask for "reveal nothing" — would return ``"****" + secret`` verbatim.
    if visible <= 0 or len(s) <= visible:
        return "*" * len(s)
    return "*" * (len(s) - visible) + s[-visible:]


def mask_head_tail(raw: str | None, *, head: int = 2, tail: int = 2) -> str | None:
    """Mask the middle, preserving ``head`` leading + ``tail`` trailing chars.

    Used for SWIFT/BIC codes, which are short enough that exposing only the
    bank+country prefix and the branch suffix keeps them unusable.

    ``ICBKCNBJSHI`` → ``IC*******HI``
    """
    if not raw:
        return raw
    s = raw.strip()
    if len(s) <= head + tail:
        return "*" * len(s)
    return s[:head] + "*" * (len(s) - head - tail) + s[-tail:]


# ── Invoice-basis (实付口径) amount redaction in audit_logs.detail ───
#
# Cloud spend exists on two bases: 折前 ``list`` (readable by everyone)
# and 实付 ``invoice`` — the net amount that reconciles with the vendor
# bill, readable only inside the finance / cloud_admin / global_admin
# circle (``app.domains.cmdb.cost_view.can_view_invoice_measure``).
# Handing the same reader a list figure and an invoice figure for the
# same (scope, period) discloses the company's negotiated discount rate.
#
# ``audit_logs.detail`` carries invoice-basis amounts to a WIDER audience
# than the cost APIs do: ``can_access_audit`` admits ``auditor`` +
# ``management``, both outside the circle, and the cloud bill-alert
# writers persist the evaluated amounts into ``detail`` for forensics
# ("why did this card fire at 3am"). Narrowing the audit gate is the
# wrong fix — an auditor seeing WHY an alert fired IS their job — so the
# audit READ path redacts per row and the stored row stays complete.
#
# What must go: absolute amounts, and anything that reconstructs one (a
# native-currency amount plus its FX rate reconstructs the USD figure).
# What may stay: dimensionless same-basis ratios — a multiplier over two
# invoice-basis figures cancels the discount and discloses nothing.

# Marker the WRITER stamps into ``detail`` to record which basis the row
# was evaluated on. Absent ⇒ the reader cannot prove the row is list
# basis and redacts (see ``redact_invoice_basis_amounts``).
AUDIT_COST_BASIS_KEY = "cost_basis"
COST_BASIS_LIST = "list"
COST_BASIS_INVOICE = "invoice"

# Classification of every ``audit_logs.action`` known to carry money in
# ``detail``. An action MISSING from this table is treated as
# invoice-basis: that is the fail-safe branch, and the only way a new
# cost-bearing audit action can reach a non-circle reader is by being
# classified here on purpose. Keep the reason on each line — the next
# reader has to be able to check the call site, not trust the label.
#
# ``COST_BASIS_PER_ROW``  basis is recorded per row under
#                         ``AUDIT_COST_BASIS_KEY``; rows without the
#                         marker redact.
# ``COST_BASIS_NON_CLOUD`` money that has no list/invoice duality at all
#                         (asset / procurement / finance budget / AI-key
#                         spend) — nothing to reconstruct, passes through.
COST_BASIS_PER_ROW = "per_row"
COST_BASIS_NON_CLOUD = "non_cloud"

AUDIT_ACTION_COST_BASIS: dict[str, str] = {
    # Cloud bill spike / threshold alerts. The spike evaluator picks its
    # basis per tick (``spike_measure`` = list only when today AND every
    # baseline day carry a list value, else net) so the basis is a
    # property of the ROW, not of the action.
    # [lifecycle_service.py::_audit_cloud_scope_alert_fired]
    "cloud.scope_alert.fired": COST_BASIS_PER_ROW,
    # Monthly integration budget alert reads
    # ``cmdb_billing_monthly.total_cost_usd`` — that table has no list
    # column, so the figure is always the provider's net amount.
    # [billing_alerts.py::_record_alert_audit]
    "cmdb.billing_alert_sent": COST_BASIS_INVOICE,
    # Owner-confirmation prompts sum ``fact_engineering_amount()``
    # (= COALESCE(list, invoice)) — engineering basis by construction,
    # identical to what every research surface already shows.
    # [owner_confirmation_service.py]
    "cloud.scope.owner_semiannual_prompt": COST_BASIS_LIST,
    "cloud.scope.owner_zero_spend_prompt": COST_BASIS_LIST,
    # Requester-supplied budget ask on a new cloud scope — a declared
    # ceiling, never a billed amount. [scope_request_service.py]
    "cloud.scope_request.submit": COST_BASIS_NON_CLOUD,
    # AI-key spend vs budget — the AI-spending domain has a single
    # measure; no cloud discount is inferable from it.
    # [apikey/budget_alerts.py]
    "apikey.budget_alert_sent": COST_BASIS_NON_CLOUD,
    # Asset / procurement / software-seat / finance-budget money.
    "software_addon.request": COST_BASIS_NON_CLOUD,
    "software_addon.admin_approve": COST_BASIS_NON_CLOUD,
    "software_addon.approved": COST_BASIS_NON_CLOUD,
    "software_billing.close": COST_BASIS_NON_CLOUD,
    "asset.dispose.auto_via_approval": COST_BASIS_NON_CLOUD,
    "finance.budget.create": COST_BASIS_NON_CLOUD,
    "finance.budget.update": COST_BASIS_NON_CLOUD,
    # White-hat bounty money. A reward is ONE agreed USD figure — there is
    # no list-vs-invoice duality to reconstruct a cloud discount from, and
    # the amount is something we told the researcher in writing. Left
    # unclassified these fell through to the invoice-basis fail-safe, so an
    # auditor or management reading the audit trail saw '[redacted]' where
    # the reward should be — on the exact three actions that record money
    # decisions. [bounty/router.py::set_reward / set_payment / create_tier]
    "bounty.report.reward": COST_BASIS_NON_CLOUD,
    "bounty.report.payment": COST_BASIS_NON_CLOUD,
    "bounty.tier.create": COST_BASIS_NON_CLOUD,
    # 2026-07-30 sweep round 2 — every remaining audit writer whose
    # detail carries a money-SHAPED key (AST scan of all `.log(...,
    # detail=...)` call sites). All non-cloud: single-measure figures
    # with no list/invoice duality, so nothing reconstructs a discount.
    # `price` on a deleted license plan. [asset/router.py]
    "asset.license_plan.delete": COST_BASIS_NON_CLOUD,
    # `unit_price` on a software add-on catalogue item. [asset/router.py]
    "software_addon.item_create": COST_BASIS_NON_CLOUD,
    # `from_price` / `to_price` on a seat-plan change. [asset/router.py]
    "software_seat.plan_changed": COST_BASIS_NON_CLOUD,
    # `monthly_cost` / `annual_cost` — telecom dedicated-line contract
    # price, no list column exists. [cmdb/router.py]
    "cmdb.dedicated_line.create": COST_BASIS_NON_CLOUD,
    # `usd_per_unit` — a pinned MARKET FX rate, public by nature.
    # [finance/router.py]
    "finance.fx.pin": COST_BASIS_NON_CLOUD,
    # `residual_rate` — a depreciation percentage, not money at all;
    # the `rate` shape token would eat it. [asset/router.py]
    "asset.depreciation_policy.upsert": COST_BASIS_NON_CLOUD,
    # `balance_legal` / `balance_welfare` — annual-leave DAY counts;
    # the `balance` shape token would eat them. [hr_tools/service.py]
    "hr_tools.annual_leave.calculate": COST_BASIS_NON_CLOUD,
    # `mapped_cost_type` — a label, not an amount. [finance/router.py]
    "finance.netsuite.map_account_cost_type": COST_BASIS_NON_CLOUD,
    # `assign_cost_center` — a cost-centre NAME. [cmdb/cost_router.py]
    "cost.allocation_rule.created": COST_BASIS_NON_CLOUD,
    # `charge_user_id` — a UUID; the `charge` shape token would eat it.
    # [approval/service.py]
    "software_addon.approval_refused_ownership_mismatch": COST_BASIS_NON_CLOUD,
}

# Key-name shapes that carry an absolute money figure (or an FX rate,
# which reconstructs one). Matched on whole underscore-delimited tokens
# so ``multiplier`` / ``recipients_count`` / ``period_key`` survive while
# ``current_usd`` / ``baseline_usd`` / ``fx_rate_cny_per_usd`` /
# ``total_amount`` / ``cost_by_service`` do not. Deliberately WIDER than
# the keys today's writers emit — an unclassified action's new money key
# must be caught by shape, not by someone remembering to list it.
_MONEY_KEY_RE = re.compile(
    r"(?:^|_)(?:"
    r"usd|cny|jpy|eur|hkd|amount|amounts|cost|costs|spend|budget|"
    r"price|prices|fee|fees|fx|rate|rates|balance|revenue|charge|charges"
    r")(?:_|$)",
    re.IGNORECASE,
)

REDACTED_AMOUNT = "[redacted]"
REDACTED_REASON = "invoice_basis_cost_measure"

# Bookkeeping keys the redactor must leave alone: the basis marker is
# metadata (redacting it would hide WHY the row was redacted), the two
# output keys must not be re-matched on a second pass, and the write
# side's withhold markers (#2097: `amounts_withheld` = "the amounts are
# absent BY POLICY, not because the cron failed") are boolean/enum
# policy metadata whose entire purpose is to be readable by the
# OUT-OF-CIRCLE audience — the `amounts` shape token must not eat them.
_REDACTION_EXEMPT_KEYS = frozenset({
    AUDIT_COST_BASIS_KEY, "_redacted_fields", "_redacted_reason",
    "amounts_withheld", "amounts_withheld_reason",
})


def is_money_detail_key(key: str) -> bool:
    """Does ``key`` name an absolute money figure (or an FX rate)?"""
    if key in _REDACTION_EXEMPT_KEYS:
        return False
    return bool(_MONEY_KEY_RE.search(key))


def _redact_money_tree(obj: Any, hits: list[str], prefix: str = "") -> Any:
    """Recursively replace money-keyed values with ``REDACTED_AMOUNT``.

    A money-named key is replaced WHOLE regardless of value type — a
    ``cost_by_service`` map is entirely amounts, so descending into it
    and matching per-key would leak every service line.
    """
    if isinstance(obj, dict):
        out: dict[str, Any] = {}
        for k, v in obj.items():
            path = f"{prefix}{k}"
            if isinstance(k, str) and is_money_detail_key(k):
                hits.append(path)
                out[k] = REDACTED_AMOUNT
            else:
                out[k] = _redact_money_tree(v, hits, f"{path}.")
        return out
    if isinstance(obj, list):
        return [
            _redact_money_tree(v, hits, f"{prefix}{i}.")
            for i, v in enumerate(obj)
        ]
    return obj


# ── Prose-money redaction (notification bodies) ─────────────────────
#
# ``notification_outbox.body_text`` is PROSE, not JSON — the cloud
# billing alert cards interpolate amounts as ``$12,345.67 USD`` /
# ``¥1,234`` style fragments. The admin outbox read surface
# (``require_admin_read`` = global_admin + auditor + management) is
# wider than the invoice circle, so a circle-recipient's threshold card
# body (which legitimately carries the absolute amounts) must not be
# re-readable verbatim by an out-of-circle admin reader — the sweep
# doc's 表四 ``body_text`` row, fix option (d).
#
# Matches a currency symbol followed by a number, or a number followed
# by an ISO-ish currency word. Deliberately does NOT match bare numbers
# (period keys, rule ordinals, multipliers like ``6.67×`` are the
# signal the card exists to deliver and reconstruct nothing).
_MONEY_TEXT_RE = re.compile(
    r"[$€£¥￥]\s?\d[\d,]*(?:\.\d+)?"
    r"(?:\s?(?:USD|CNY|RMB|JPY|EUR|HKD))?"
    r"|(?<![\w.,])\d[\d,]*(?:\.\d+)?\s?(?:USD|CNY|RMB|JPY|EUR|HKD)\b",
    re.IGNORECASE,
)
REDACTED_TEXT_AMOUNT = "[金额已保密 amount withheld]"


def redact_money_in_text(text: str | None) -> tuple[str | None, bool]:
    """Replace currency-amount fragments in prose with a marker.

    Returns ``(new_text, hit)``; ``hit`` is False when nothing matched
    so callers can serve the original object untouched. Falsy input
    passes through unchanged (same convention as the maskers above).
    """
    if not text:
        return text, False
    new, n = _MONEY_TEXT_RE.subn(REDACTED_TEXT_AMOUNT, text)
    return (new, True) if n else (text, False)


def redact_invoice_basis_amounts(
    detail: dict[str, Any] | None,
    *,
    action: str | None,
    can_view_invoice: bool,
) -> dict[str, Any] | None:
    """Strip invoice-basis amounts from an ``audit_logs.detail`` payload.

    ``can_view_invoice`` is
    ``cost_view.can_view_invoice_measure(reader_roles)`` — inside the
    circle the row passes through untouched, so the DB row and the
    circle reader always see the same thing.

    Outside the circle the action's classification in
    ``AUDIT_ACTION_COST_BASIS`` decides. List-basis and non-cloud money
    passes through in full (over-redacting the common case makes the
    audit trail useless); invoice-basis, per-row-marked-but-unproven,
    and UNKNOWN actions get every money-shaped key replaced. Two
    bookkeeping keys are appended so the reader can tell a redacted row
    from a row that never carried amounts.
    """
    if not detail or can_view_invoice:
        return detail
    basis = AUDIT_ACTION_COST_BASIS.get(action or "")
    if basis in (COST_BASIS_LIST, COST_BASIS_NON_CLOUD):
        return detail
    if basis == COST_BASIS_PER_ROW:
        marker = detail.get(AUDIT_COST_BASIS_KEY)
        if isinstance(marker, str) and marker.strip().lower() == COST_BASIS_LIST:
            return detail
        # Marker missing (every row written before 2026-07-29) or net —
        # we cannot PROVE list basis, so redact. Costs the auditor the
        # amounts on historical list-basis spike rows; the alternative
        # leaks the discount rate on every historical net-basis row.
    hits: list[str] = []
    redacted = _redact_money_tree(detail, hits)
    if not hits:
        return detail
    redacted["_redacted_fields"] = hits
    redacted["_redacted_reason"] = REDACTED_REASON
    return redacted