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

"""业务事实的**命名构造**：充值 / 兑换 / 积分到账 / 绑定。

## 为什么单独一个模块

`emit()` 只负责「往时间线写一条」，它不知道「一次兑换」到底该长什么样。
以前这件事由收集器从 `douyu_tasks.result` 的 JSON 里**反推**（猜字段路径），
结果是：

- 新增业务类型没人想起改 allowlist → **静默丢数据**
  （电竞兑换、和平小店兑换、锁单支付、电竞礼物积分从来没进过时间线）；
- 同一件事的字段口径散在 runner 与收集器两处，容易分叉；
- `success` 不等于「业务真的发生了」时，把假成功也记成事实。

现在反过来：**业务代码在动作成功的那一刻，用这里的构造函数声明事实**。
每类事实的字段、标题、金额符号只定义一次。

## 两条硬规则

1. **必须紧贴 `_mark_task(..., "success", ...)` 之前调用。**
   `_mark_task` 内部 `db.commit()`，而 `emit()` 用 `db.begin_nested()` 只 RELEASE
   SAVEPOINT、不提交外层事务；`douyu_runner` 的 `finally` 只 `close()`（未提交即回滚）。
   在 `_mark_task` **之后**发事件会被静默回滚 —— 而且 `emit()` 返回 True、无任何日志。
   放在之前还有个好处：事件与任务状态**同一次提交**，原子。

2. **幂等键用 `task_ref(task.id)`**，与旧收集器的 key 完全一致。
   于是迁移窗口内「直发」与「仍在跑的收集器」会自动去重，可以安全并存后再删作业。
   ⚠️ 不要用秒级时间戳当幂等键：撞唯一键会让 `emit()` 静默吞掉事件
   （见 `account_lifecycle` 里那段复盘）。
"""

from __future__ import annotations

from datetime import datetime
from typing import Any

from sqlalchemy.orm import Session

from ..models import Account
from .account_events_service import (
    CATEGORY_BUSINESS,
    EVENT_BIND_OK,
    EVENT_EXCHANGE,
    EVENT_GOLD_SPENT,
    EVENT_POINTS_CREDIT,
    EVENT_POINTS_EXPIRE,
    EVENT_RECHARGE,
    OFFICIAL_SYNC_SOURCE,
    emit,
    to_naive_utc,
)


def task_ref(task_id: int, kind: str = "") -> str:
    """业务事件的幂等键：一条任务行对应一条事实。

    与旧收集器（`f"douyu_task:{row['id']}"`）**完全一致**，所以直发与收集器
    同时观测到同一笔业务时只会写一条 —— 这让「先直发、观察一段、再删收集器」
    成为安全的迁移路径。
    """
    base = f"douyu_task:{int(task_id)}"
    # 同一个任务可能产生**多条**事件（例如送礼：既「积分到账」又「消耗鱼翅」）。
    # 它们共用 task_ref 会撞唯一键、后写的那条被静默丢弃 —— 所以副事件带一个后缀。
    # 不带后缀的形态与旧收集器一致，主事件继续沿用它。
    return f"{base}:{kind}" if kind else base


def task_actor(task: Any) -> dict[str, Any]:
    """业务事件的「谁做的」—— 发起这条任务的用户。

    为什么用 `task.created_by`：任务由客服在界面上发起（池路径下也就是**租用该账号
    的客服**），所以它就是操作人。以前所有业务事件（充值/兑换/开通/积分）都不带这个
    字段，生产 34,462 条事件里有 24,502 条（71%）查不出责任人 —— 池化之后
    「客服租用期间做了什么」是必须能回答的，所以补齐。

    调用处一律用 `**task_actor(task)`，默认值仍是「无操作人」，
    因此老调用方一行不改也不会变。
    """
    created_by = getattr(task, "created_by", None)
    if not created_by:
        return {"actor_type": "system", "actor_id": None}
    return {"actor_type": "user", "actor_id": int(created_by)}


def record_recharge(
    db: Session,
    account: Account,
    *,
    task_id: int,
    amount: int | None,
    occurred_at: datetime | None = None,
    order_id: str = "",
    supplier_order_id: str = "",
    channel: str = "",
    douyu_uid: str = "",
    balance_after: int | None = None,
    confirmed: bool = True,
    actor_type: str = "",
    actor_id: int | None = None,
) -> bool:
    """充值鱼翅到账。

    `amount` 用**本地已知的充值金额**，不要从 `result` 里找 —— 二维码链路的
    `result` 里根本没有金额（`create_gold_qr` 只返回 `pay_url/token/raw`），
    旧收集器因此在二维码充值上恒写 `amount=None`，导致 `accounts.recharge_total`
    系统性少计。

    `confirmed=False` 表示只有供应商的终态回执、没有核对斗鱼余额变化。
    """
    title = f"充值 {amount}" if amount is not None else "充值到账"
    if not confirmed:
        title += "（供应商回执）"
    return emit(
        db,
        account_id=account.id,
        event_type=EVENT_RECHARGE,
        title=title,
        category=CATEGORY_BUSINESS,
        occurred_at=occurred_at,
        amount=amount,
        balance_after=balance_after,
        detail={
            "order_id": order_id,
            "supplier_order_id": supplier_order_id,
            "channel": channel,
            "douyu_uid": douyu_uid,
            "confirmed": confirmed,
        },
        source="douyu_task",
        subject_type="recharge",
        subject_key=order_id or supplier_order_id or "",
        source_ref=task_ref(task_id),
        actor_type=actor_type,
        actor_id=actor_id,
    )


def record_exchange(
    db: Session,
    account: Account,
    *,
    task_id: int,
    name: str,
    score: int | None,
    occurred_at: datetime | None = None,
    commodity_id: str = "",
    order_id: str = "",
    exchange_id: str = "",
    game_name: str = "",
    game_channel: str = "",
    points_after: int | None = None,
    scope: str = "elite",
    actor_type: str = "",
    actor_id: int | None = None,
) -> bool:
    """商品/皮肤兑换成功 —— **只在真的消耗了积分并兑换成功时调用**。

    注意「预约到货 / 已预兑 / 预兑确认」这三类也把任务标成 `success`，但它们
    **不消耗积分、也没兑换到东西**，绝不能走这里（旧收集器只按 `status=="success"`
    过滤，于是每预约一次就生成一条金额为空的 `exchange`，把 `exchange_count` 灌大）。

    `score` 为负写入 `amount`（消耗），`points_after` 是兑换后的积分余额。
    """
    return emit(
        db,
        account_id=account.id,
        event_type=EVENT_EXCHANGE,
        title=f"兑换 {name}" if name else "兑换",
        category=CATEGORY_BUSINESS,
        occurred_at=occurred_at,
        amount=-int(score) if isinstance(score, (int, float)) else None,
        balance_after=points_after,
        detail={
            "commodity_id": commodity_id,
            "commodity_name": name,
            "order_id": order_id,
            "exchange_id": exchange_id,
            "game_name": game_name,
            "game_channel": game_channel,
            "score": score,
            "scope": scope,
        },
        source="douyu_task",
        subject_type="commodity",
        subject_key=commodity_id or name or "",
        source_ref=task_ref(task_id),
        actor_type=actor_type,
        actor_id=actor_id,
    )


def record_gold_spent(
    db: Session,
    account: Account,
    *,
    task_id: int,
    amount: int | None,
    occurred_at: datetime | None = None,
    reason: str = "",
    source_label: str = "",
    balance_after: int | None = None,
    confirmed: bool = True,
    actor_type: str = "",
    actor_id: int | None = None,
) -> bool:
    """鱼翅消耗（送礼等）。

    `amount` 传**正数**，这里写成负号（与 `exchange` 一样，出项用负数）。
    金额取**实测的余额差**（送礼前后各刷一次鱼翅），不是按礼物单价推算 ——
    单价会变、活动会打折，只有观测到的差是真的。

    余额差拿不到（或没有减少）时**不发事件**：宁可少一条，也不要记一笔没发生的消耗。
    """
    if not amount or int(amount) <= 0:
        return False
    spent = int(amount)
    title = f"{source_label}消耗鱼翅 {spent}" if source_label else f"消耗鱼翅 {spent}"
    if not confirmed:
        title += "（未确认扣减）"
    return emit(
        db,
        account_id=account.id,
        event_type=EVENT_GOLD_SPENT,
        title=title,
        category=CATEGORY_BUSINESS,
        occurred_at=occurred_at,
        amount=-spent,
        balance_after=balance_after,
        detail={"reason": reason, "source": source_label, "confirmed": confirmed},
        source="douyu_task",
        subject_type="gold",
        # 副事件：同一个任务还会发「积分到账」，所以带后缀避免撞幂等键
        source_ref=task_ref(task_id, "gold"),
        actor_type=actor_type,
        actor_id=actor_id,
    )


def record_points_expire(
    db: Session,
    account: Account,
    *,
    amount: int | None,
    occurred_at: datetime | None = None,
    expires_at: Any = None,
    confirmed: bool = False,
    actor_type: str = "",
    actor_id: int | None = None,
) -> bool:
    """积分到期（资产减少）。

    ⚠️ **默认 `confirmed=False`**：斗鱼只在积分查询里给一个 `willExpireTime`
    （到期时刻），到期**实际扣了多少分**我们并没有亲眼看到。
    所以默认只记「预计到期」，并在 detail 里标清楚；要记成已确认，
    必须由重新查询后真正观测到余额下降的调用方显式传 `confirmed=True`。
    """
    title = f"积分到期 -{amount}" if amount else "积分到期"
    if not confirmed:
        title += "（预计）"
    return emit(
        db,
        account_id=account.id,
        event_type=EVENT_POINTS_EXPIRE,
        title=title,
        category=CATEGORY_BUSINESS,
        occurred_at=occurred_at,
        amount=-int(amount) if amount else None,
        detail={"confirmed": confirmed, "expires_at": expires_at},
        source="points_expire",
        subject_type="points",
        actor_type=actor_type,
        actor_id=actor_id,
    )


def record_points_credit(
    db: Session,
    account: Account,
    *,
    task_id: int,
    gain: int | None,
    occurred_at: datetime | None = None,
    points_after: int | None = None,
    expire_at: Any = None,
    confirmed: bool = True,
    source_label: str = "精英令",
    actor_type: str = "",
    actor_id: int | None = None,
) -> bool:
    """积分到账（送礼/任务发放）。

    `confirmed` 必须如实反映「有没有真的看到积分涨上去」。送礼任务在**没确认**
    到账时也会把任务标成 `success`（消息里加一句「（未确认涨到 X）」），旧收集器
    照样把 `gift_points_expected_gain` 当成事实写进台账 —— 把「预期」记成了「已到账」。

    这里保留事件（延迟到账也是真实发生的发放），但**把 `confirmed` 明确落进
    detail**：宁可标清楚是未确认，也不假装它是确认过的数字。
    """
    title = f"积分到账 +{gain}" if gain is not None else "积分到账"
    title = f"{source_label}{title}"
    if not confirmed:
        title += "（未确认）"
    return emit(
        db,
        account_id=account.id,
        event_type=EVENT_POINTS_CREDIT,
        title=title,
        category=CATEGORY_BUSINESS,
        occurred_at=occurred_at,
        amount=int(gain) if isinstance(gain, (int, float)) else None,
        balance_after=points_after,
        detail={
            "confirmed": confirmed,
            "expire_at": expire_at,
            "source": source_label,
        },
        source="douyu_task",
        subject_type="points",
        source_ref=task_ref(task_id),
        actor_type=actor_type,
        actor_id=actor_id,
    )


def record_bind(
    db: Session,
    account: Account,
    *,
    task_id: int,
    scope: str,
    role_name: str = "",
    occurred_at: datetime | None = None,
    act_alias: str = "",
    rebind: bool = False,
    weak: bool = False,
    actor_type: str = "",
    actor_id: int | None = None,
) -> bool:
    """角色绑定 / 换绑**真的发生了**。

    ⚠️ 只在「确认接口成功」时调用。`confirm_bind` 里还有一条分支是
    「检测到账号**本来就已经绑定**」——它什么都没做也把任务标成 `success`，
    旧收集器把每次重复点「确认绑定」都记成一条 `bind_ok`。

    `weak=True` 表示确认接口成功、但回查拿不到结果（没能亲眼确认生效），
    仍然记，但标出来。
    """
    label = "换绑" if rebind else "绑定"
    scope_label = {"elite": "精英宝典", "esports": "电竞手册", "peace": "和平小店"}.get(
        scope, scope
    )
    title = f"{scope_label}{label}"
    if role_name:
        title += f"：{role_name}"
    if weak:
        title += "（未回查确认）"
    return emit(
        db,
        account_id=account.id,
        event_type=EVENT_BIND_OK,
        title=title,
        category=CATEGORY_BUSINESS,
        occurred_at=occurred_at,
        detail={
            "scope": scope,
            "role_name": role_name,
            "act_alias": act_alias,
            "rebind": rebind,
            "confirmed": not weak,
        },
        source="douyu_task",
        subject_type=f"{scope}_role",
        subject_key=role_name or "",
        source_ref=task_ref(task_id),
        actor_type=actor_type,
        actor_id=actor_id,
    )


def record_official_exchange(
    db: Session,
    account: Account,
    *,
    commodity_id: str,
    commodity_name: str,
    score: int | None,
    occurred_at: datetime | None,
    show_name: str = "",
    exchange_num: int | None = None,
    scope: str = "elite",
    actor_type: str = "",
    actor_id: int | None = None,
) -> bool:
    """把斗鱼官方兑换记录写成账号事件。

    ## 为什么要写进事件流

    列表的「兑换记录」列一直来自我们自己的 `account_events`，而事件是**上线后**
    才开始记的 —— 上线前的兑换在界面上就是空白，看着像"从没兑换过"
    （用户报过：账号明明有官方兑换记录，列表却显示「未兑换」）。

    官方记录**带真实兑换时间**（`time` 字段），所以可以直接补成正常的历史事件，
    时间线因此变完整（这就是"账号全生命周期"该有的样子），
    列表的聚合也随之自动准确 —— 不需要再单独维护一套"官方条数"的旁路。

    ## 幂等

    `source_ref` 用 `官方:账号:商品:时刻`，同一条记录查多少次都只落一条。
    这也是必须的：一键查询会被反复执行。
    """
    if not commodity_id:
        return False
    return emit(
        db,
        account_id=account.id,
        event_type=EVENT_EXCHANGE,
        title=f"兑换 {commodity_name}" if commodity_name else "兑换",
        category=CATEGORY_BUSINESS,
        occurred_at=occurred_at,
        amount=-int(score) if score else None,
        detail={
            "commodity_id": commodity_id,
            "commodity_name": commodity_name,
            "score": score,
            "exchange_num": exchange_num,
            "show_name": show_name,
            # 标明来源：这条是**从官方记录补录**的，不是我们亲眼看着兑换的
            "source": OFFICIAL_SYNC_SOURCE,
            # 哪个作用域的官方记录（精英宝典 / 电竞手册 / 和平小店）
            "scope": scope,
        },
        source=OFFICIAL_SYNC_SOURCE,
        # 台账检索键必须是**斗鱼商品 ID**：365 天规则与 `ix_account_events_ledger`
        # 都按 `subject_key` 判「该账号该商品最近一次兑换」。
        #
        # 2026-10-07 修：这里以前漏传 `subject_key`（只设了 subject_type="goods"），
        # 于是 10,612 条官方补录事件的主键全是空串 —— 用 subject_key 查历史会漏掉
        # 87% 的记录，365 天规则形同虚设。`subject_type` 同时统一为 commodity，
        # 与 `record_exchange`（我方任务直发）保持一致。
        # 幂等靠下面的 `source_ref`，与 subject_key 无关，改这里不会造成重复。
        subject_type="commodity",
        subject_key=commodity_id,
        source_ref=(
            f"official_exchange:{scope}:{account.id}:{commodity_id}:{_stamp(occurred_at)}"
        ),
        actor_type=actor_type,
        actor_id=actor_id,
    )


def _stamp(value: datetime | None) -> str:
    """幂等键里的时间部分；没有时间就用固定标记（同一商品仍只落一条）。"""
    return to_naive_utc(value).strftime("%Y%m%d%H%M%S") if value else "unknown"