live-hub-py/web/backend/services/account_binding.py

"""角色绑定状态：把三个「垃圾场」列收敛成明确枚举。

## 背景：`bind_status` 崩掉的方式

历史上账号的绑定状态存在**一个自由文本列** `accounts.bind_status` 里，写入方遍布
6 个文件、29 处，取值 20 多种，而且**语义混杂**：

- 真正的状态：`bind_confirmed` / `game_not_bound`
- 动作痕迹：`bind_qr_generated` / `goods_locked` / `elite_opened`
- 纯查询痕迹：`points_queried` / `change_time_queried` / `csrf_token_ready`

因为它是**覆盖式**的，`bind_confirmed` 会被之后任意一次查询（例如
`limited_goods_queried`）盖掉 —— 于是「这个账号到底绑没绑」查不出来。
`esports_bind_status` / `xpd_bind_status` 是同一个病。

## 收敛规则（设计文档 §3.6）

1. **状态列只放「现在是什么」**：`unknown` / `unbound` / `pending` / `bound`；
2. **动作与查询痕迹一律不再写状态列** —— 动作已经进 `account_events`
   （S5 的直发事件），查询痕迹本来就不该是状态；
3. 三列各管一个作用域，**一列一义**：

   | 列 | 作用域 |
   | --- | --- |
   | `accounts.role_bind_status` | 精英宝典角色绑定 |
   | `accounts.esports_bind_status` | 电竞手册角色绑定 |
   | `accounts.xpd_bind_status` | 和平小店角色绑定 |

> ⚠️ `pending` 的含义是「角色已就绪 / 二维码已出，但**还没确认生效**」。
> 斗鱼侧的疑点宁可为 `unknown`，也不要冒充 `bound`。
"""

from __future__ import annotations

from ..models import Account

# ── 作用域 ──────────────────────────────────────────────
SCOPE_ELITE = "elite"
SCOPE_ESPORTS = "esports"
SCOPE_PEACE = "peace"
ALL_SCOPES = (SCOPE_ELITE, SCOPE_ESPORTS, SCOPE_PEACE)

# ── 取值域（**只有这四个**）────────────────────────────
BIND_UNKNOWN = "unknown"  # 还没查过 / 旧数据不可信
BIND_UNBOUND = "unbound"  # 明确没绑定
BIND_PENDING = "pending"  # 角色就绪或二维码已出，尚未确认生效
BIND_BOUND = "bound"  # 确认生效

ALL_BIND_STATES = (BIND_UNKNOWN, BIND_UNBOUND, BIND_PENDING, BIND_BOUND)

BIND_LABELS = {
    BIND_UNKNOWN: "未知",
    BIND_UNBOUND: "未绑定",
    BIND_PENDING: "待确认",
    BIND_BOUND: "已绑定",
}

#: 作用域 → `accounts` 上的列名
SCOPE_COLUMNS = {
    SCOPE_ELITE: "role_bind_status",
    SCOPE_ESPORTS: "esports_bind_status",
    SCOPE_PEACE: "xpd_bind_status",
}

SCOPE_LABELS = {
    SCOPE_ELITE: "精英宝典角色",
    SCOPE_ESPORTS: "电竞手册角色",
    SCOPE_PEACE: "和平小店角色",
}

# ── 旧自由文本 → 新枚举 ─────────────────────────────────
#
# 只映射**无歧义**的取值；其余一律 `unknown`。判定标准是「这个旧值是不是
# 明确陈述了绑定关系」：
#   - `bind_confirmed` 是「确认绑定成功」那一刻写的，而绑定不会被之后的查询撤销
#     → 可以确信是 bound；
#   - `game_not_bound` 明确说没绑 → unbound；
#   - `bind_qr_generated` / `xpd_bind_qr_ready` 只是出了二维码 → pending；
#   - `game_queried` / `points_queried` / `goods_locked` 之类**什么都没说**
#     → unknown（设计文档 §5.5：不猜测性回填）。
#
# ⚠️ 这三张表**必须与迁移 `20261004_0044` 里的 CASE 逐字对应**，有一条用例
# 真跑那段 SQL 与它们比对（分叉了不会报错，只会让回填结果与运行时不一致）。

LEGACY_ROLE_BIND_MAP: dict[str, str] = {
    "bind_confirmed": BIND_BOUND,
    "bind_qr_generated": BIND_PENDING,
    "game_not_bound": BIND_UNBOUND,
}

LEGACY_ESPORTS_BIND_MAP: dict[str, str] = {
    "esports_bound": BIND_BOUND,
    "esports_bind_ready": BIND_PENDING,
    "esports_bind_confirming": BIND_PENDING,
    "esports_qr_created": BIND_PENDING,
    "esports_role_switching": BIND_PENDING,
}

LEGACY_XPD_BIND_MAP: dict[str, str] = {
    "xpd_bound": BIND_BOUND,
    "xpd_bind_qr_ready": BIND_PENDING,
    "xpd_not_bound": BIND_UNBOUND,
}

LEGACY_MAPS = {
    SCOPE_ELITE: LEGACY_ROLE_BIND_MAP,
    SCOPE_ESPORTS: LEGACY_ESPORTS_BIND_MAP,
    SCOPE_PEACE: LEGACY_XPD_BIND_MAP,
}


def map_legacy_bind_status(scope: str, legacy: str | None) -> str:
    """旧自由文本 → 新枚举；认不出来就是 `unknown`（**不猜**）。"""
    return LEGACY_MAPS.get(scope, {}).get((legacy or "").strip(), BIND_UNKNOWN)


def get_bind_status(account: Account, scope: str) -> str:
    column = SCOPE_COLUMNS.get(scope)
    if not column:
        raise ValueError(f"未知的绑定作用域: {scope}")
    value = getattr(account, column, "") or ""
    # 兼容：迁移之前的库或测试构造的对象可能还没有新列的值
    return value if value in ALL_BIND_STATES else BIND_UNKNOWN


def set_bind_status(account: Account, scope: str, state: str) -> bool:
    """设置某个作用域的绑定状态，返回是否真的变化。

    **这是三个绑定状态列的唯一写入口**（同 `account_lifecycle` 之于 `status`）。
    `state` 必须是四个枚举值之一；传旧自由文本会直接报错，避免又把垃圾写回去。
    """
    if state not in ALL_BIND_STATES:
        raise ValueError(
            f"绑定状态只能是 {ALL_BIND_STATES} 之一，收到 {state!r}。"
            "动作/查询痕迹请进 account_events，不要写状态列。"
        )
    column = SCOPE_COLUMNS.get(scope)
    if not column:
        raise ValueError(f"未知的绑定作用域: {scope}")
    if getattr(account, column, None) == state:
        return False
    setattr(account, column, state)
    return True


def bind_payload(account: Account) -> dict[str, str]:
    """给接口/前端用的三列快照（值 + 中文标签）。"""
    payload: dict[str, str] = {}
    for scope in ALL_SCOPES:
        state = get_bind_status(account, scope)
        payload[f"{scope}_bind_status"] = state
        payload[f"{scope}_bind_label"] = BIND_LABELS[state]
    return payload