yunke/yunke-server/app/api/routers/staff_chat_stats.py

"""GET /api/staff-chat-stats — 微信员工 5 数字 + 客户明细（本地 message 表实时聚合）。

数据全本地：按 员工微信 wxid × ms-epoch 日期窗 聚合 1:1 私聊（room_id 过滤），
不访问云客。5 汇总数从 customers 在 Python 侧派生（spec §5）。
跨库日期提取（codebase 首例）镜像 app/storage/repos.py 的方言检测写法。
"""
from __future__ import annotations

import time
from typing import Annotated

import sqlalchemy as sa
from fastapi import APIRouter, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession

from app.api.deps import get_session
from app.api.schemas import StaffChatCustomerOut, StaffChatStatsOut, StaffChatStatsSummary
from app.db.models import Customer, Message

router = APIRouter(prefix="/api", tags=["staff-chat-stats"])

SessionDep = Annotated[AsyncSession, Depends(get_session)]

_MS_PER_DAY = 86_400_000


def _private_window(msg, owner: str, start: int, end: int):
    """1:1 私聊 + 窗口过滤条件（channel 固定 wechat），排除 timestamp IS NULL。"""
    return [
        msg.channel == "wechat",
        msg.owner_account_id == owner,
        sa.or_(msg.room_id.is_(None), msg.room_id == ""),
        msg.timestamp.is_not(None),
        msg.timestamp >= start,
        msg.timestamp <= end,
    ]


async def _aggregate(session: AsyncSession, owner: str, start: int, end: int) -> tuple[list, dict[str, bool | None]]:
    """返回 [(talker_id, send, recv, interaction_days, last_time)] + last_is_send map。"""
    # 跨库日期提取：从 ms-epoch 取日期。镜像 repos.py 方言检测。
    dialect = session.bind.name  # 镜像 app/storage/repos.py（.name 等价 .dialect.name）
    if dialect == "sqlite":
        day_expr = sa.func.date(Message.timestamp / 1000, "unixepoch")
    else:  # mysql（生产）
        day_expr = sa.func.date(sa.func.from_unixtime(Message.timestamp / 1000))

    base = _private_window(Message, owner, start, end)
    agg_stmt = (
        sa.select(
            Message.talker_id,
            sa.func.sum(sa.case((Message.is_send.is_(True), 1), else_=0)).label("send_count"),
            sa.func.sum(sa.case((Message.is_send.is_(False), 1), else_=0)).label("receive_count"),
            sa.func.count(sa.func.distinct(day_expr)).label("interaction_days"),
            sa.func.max(Message.timestamp).label("last_time"),
        )
        .where(*base)
        .group_by(Message.talker_id)
        .order_by(Message.talker_id)
    )
    agg_rows = (await session.execute(agg_stmt)).all()

    # 每客户最后一条消息方向：ROW_NUMBER 窗口取 rn=1（tie-break msg_id 保证确定性）
    inner = (
        sa.select(
            Message.talker_id,
            Message.is_send,
            sa.func.row_number()
            .over(
                partition_by=Message.talker_id,
                order_by=(Message.timestamp.desc(), Message.msg_id.desc()),
            )
            .label("rn"),
        )
        .where(*base)
    )
    sub = inner.subquery("last_msg")
    last_stmt = sa.select(sub.c.talker_id, sub.c.is_send).where(sub.c.rn == 1)
    last_map = {tid: is_send for tid, is_send in (await session.execute(last_stmt)).all()}

    return agg_rows, last_map


async def _customer_names(session: AsyncSession, owner: str, talker_ids: list[str]) -> dict[str, tuple[str, str, str]]:
    if not talker_ids:
        return {}
    stmt = sa.select(
        Customer.friend_id,
        Customer.friend_remark,
        Customer.friend_nickname,
        Customer.real_name,
        Customer.head_image_url,
    ).where(
        Customer.channel == "wechat",
        Customer.owner_account_id == owner,
        Customer.friend_id.in_(talker_ids),
    )
    info: dict[str, tuple[str, str, str]] = {}
    for fid, remark, nick, real, head in (await session.execute(stmt)).all():
        info[fid] = (remark or nick or real or fid, real or "", head or "")
    return info


@router.get("/staff-chat-stats")
async def get_staff_chat_stats(
    session: SessionDep,
    owner_account_id: Annotated[str, Query(min_length=1)],
    start: Annotated[int | None, Query(ge=0)] = None,
    end: Annotated[int | None, Query(ge=0)] = None,
) -> StaffChatStatsOut:
    """微信员工沟通统计：5 数字 + 客户明细（全本地 message 聚合，1:1 私聊）。"""
    if end is None:
        end = int(time.time() * 1000)
    if start is None:
        start = end - 3 * _MS_PER_DAY  # 默认近 3 天

    agg_rows, last_map = await _aggregate(session, owner_account_id, start, end)
    talker_ids = [r.talker_id for r in agg_rows]
    names = await _customer_names(session, owner_account_id, talker_ids)

    customers: list[StaffChatCustomerOut] = []
    for r in agg_rows:
        tid = r.talker_id
        last_is_send = last_map.get(tid)  # bool | None（None=最后一条方向未知）
        customers.append(
            StaffChatCustomerOut(
                talker_id=tid,
                friend_name=(names.get(tid) or (tid, "", ""))[0],
                real_name=(names.get(tid) or ("", "", ""))[1] or None,
                head_image_url=(names.get(tid) or ("", "", ""))[2] or None,
                send_count=int(r.send_count or 0),
                receive_count=int(r.receive_count or 0),
                interaction_days=int(r.interaction_days or 0),
                # True→1, False→0, None→None（方向未知不计入两率；summary 用 ==1/==0 自动排除 None）
                last_is_send=int(last_is_send) if last_is_send is not None else None,
                last_time=r.last_time,
            )
        )

    summary = StaffChatStatsSummary(
        chat_count=len(customers),
        effective_count=sum(1 for c in customers if c.send_count >= 1 and c.receive_count >= 1),
        follow_count=sum(1 for c in customers if c.interaction_days >= 2),
        last_reply_count=sum(1 for c in customers if c.last_is_send == 1),
        no_reply_count=sum(1 for c in customers if c.last_is_send == 0),
    )
    return StaffChatStatsOut(
        owner_account_id=owner_account_id,
        start=start,
        end=end,
        summary=summary,
        customers=customers,
    )