浅念
b95fc5fa6f
计数模式(无kind)返回 new_customers/new_groups/red_transfer/del_customers/
blocked_customers/voice_calls/sensitive_words(恒0,spec占位);明细模式
(?kind=)返回{kind,rows},每kind字段不同,前端按kind配列。start/end
缺省近7天。
口径(spec §4.5,全本地聚合):
- new_customers: add_time窗内;IS NULL行用该对话首条消息min(timestamp)兜底
- new_groups: chat_group.create_time窗内(复合PK按owner过滤)
- red_transfer: wechat type IN (red_packet,transfer)窗内;金额三路正则
(红包XML feedesc/转账纯文本/totalamount),实测真库转账399/399全提取
- del/blocked_customers: 现存量口径忽略窗口,明细time=updated_at毫秒
- voice_calls: phone_call.user_id存员工userId(32HEX)非wxId——先经
wechat_account.user_id换算,账号无uid返回空防跨员工串数
- voice明细字段语义对齐CallList.vue(connected=call_status==1,
media_url本地/api/media/{hash}优先OSS兜底)
真机smoke(yunke-mysql):voice 61条/新客67/删除29/拉黑1均通过;
转账金额100%提取。真库chat_group列与message.type回填滞后(运维项,另报)。
Co-Authored-By: Claude <noreply@anthropic.com>
408 lines
14 KiB
Python
408 lines
14 KiB
Python
"""GET /api/staff-stats — 员工聊天窗口 7 类计数 + 分型明细(本地聚合,spec §4.5)。
|
||
|
||
对齐云客员工聊天窗口顶部 7 个统计数字与其点击弹出的分型明细列表:
|
||
* 计数模式(无 ``kind``)→ ``{new_customers, new_groups, red_transfer,
|
||
del_customers, blocked_customers, voice_calls, sensitive_words}``
|
||
(``sensitive_words`` 恒 0——敏感词占位,spec 决策)。
|
||
* 明细模式(``?kind=``)→ ``{kind, rows}``,每 kind 字段不同(统一
|
||
``list[dict]``,前端按 kind 配置列)。
|
||
|
||
统计口径(spec §4.5,全部本地聚合,逐条实测验证):
|
||
* new_customers ``customer.add_time`` 窗内;``add_time IS NULL`` 的行用
|
||
「该对话首条消息时间」兜底(message 同 owner+talker 的
|
||
``min(timestamp)`` 在窗内即算,~38% 行靠兜底)。
|
||
* new_groups ``chat_group.create_time`` 窗内(Task 3 新列,
|
||
raw_json createTime 回填);无值行不计。
|
||
* red_transfer wechat 消息 type IN (red_packet, transfer) 且
|
||
timestamp 窗内(Task 4 已回填类型);金额从 text
|
||
红包 XML ``<feedesc>`` / 转账纯文本「发送转账：￥N」提取。
|
||
* del_customers ``delete_flag=1`` 现存量(无时间字段→忽略窗口;
|
||
明细行 time 用 updated_at 毫秒)。
|
||
* blocked_customers ``is_block=1`` 现存量(同上)。
|
||
* voice_calls ``phone_call.user_id`` 存的是员工 userId(32位HEX)而非
|
||
wxId(实测坑!)——必须先经 ``wechat_account.user_id``
|
||
换算再查,直查 wxId 恒 0。
|
||
* sensitive_words 恒 0(占位)。
|
||
|
||
start/end 缺省 = 近 7 天(对齐云客统计默认窗)。结构镜像
|
||
:mod:`app.api.routers.staff_chat_stats`(SessionDep + 每类一个私有
|
||
聚合函数 + 主端点分派)。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import time
|
||
from typing import Annotated, Any, Literal
|
||
|
||
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 StaffStatsCountsOut, StaffStatsDetailOut
|
||
from app.db.models import ChatGroup, Customer, Message, PhoneCall, WechatAccount
|
||
|
||
router = APIRouter(prefix="/api", tags=["staff-stats"])
|
||
|
||
SessionDep = Annotated[AsyncSession, Depends(get_session)]
|
||
|
||
_MS_PER_DAY = 86_400_000
|
||
|
||
KindParam = Literal[
|
||
"new_customers",
|
||
"new_groups",
|
||
"red_transfer",
|
||
"del_customers",
|
||
"blocked_customers",
|
||
"voice_calls",
|
||
]
|
||
|
||
# 红包/转账金额提取(spec §4.5 red_transfer):
|
||
# 1) 红包 XML:``<wcpayinfo><feedesc>￥88.00</feedesc>``(feedesc 内货币符号非数字)
|
||
# 2) 云客转账纯文本:``发送转账：￥100.00``
|
||
# 3) 转账 XML ``<totalamount>100.00</totalamount>`` 兜底
|
||
_AMOUNT_FEEDESC_RE = re.compile(r"<feedesc>[^0-9]*([\d.]+)")
|
||
_AMOUNT_PLAIN_RE = re.compile(r"转账[：:]?\s*[￥¥]\s*([\d.]+)")
|
||
_AMOUNT_TOTAL_RE = re.compile(r"<totalamount>\s*([\d.]+)")
|
||
|
||
|
||
def _extract_amount(text: str | None) -> float | None:
|
||
"""从消息 text 提取金额(元);双格式,失败 None。"""
|
||
if not text:
|
||
return None
|
||
m = (
|
||
_AMOUNT_FEEDESC_RE.search(text)
|
||
or _AMOUNT_PLAIN_RE.search(text)
|
||
or _AMOUNT_TOTAL_RE.search(text)
|
||
)
|
||
if not m:
|
||
return None
|
||
try:
|
||
return float(m.group(1))
|
||
except ValueError: # pragma: no cover — 正则已限定 [\d.]+,防御畸形串
|
||
return None
|
||
|
||
|
||
def _dt_ms(value: Any) -> int | None:
|
||
"""DateTime 列 → ms epoch(naive,按本地时区解释;仅作展示兜底)。"""
|
||
if value is None:
|
||
return None
|
||
return int(value.timestamp() * 1000)
|
||
|
||
|
||
async def _new_customers(
|
||
session: AsyncSession, wxid: str, start: int, end: int, detail: bool
|
||
) -> tuple[int, list[dict[str, Any]]]:
|
||
"""新客户:add_time 窗内;add_time IS NULL 用首条消息时间兜底。"""
|
||
base = [Customer.channel == "wechat", Customer.owner_account_id == wxid]
|
||
# 兜底口径:每个 add_time 为空的客户,取其与该员工首条消息 min(timestamp)
|
||
# (LEFT JOIN 走 ix_message_channel_owner_talker_time,HAVING 在 SQL 侧过滤窗)。
|
||
fallback = (
|
||
sa.select(
|
||
Customer.friend_id,
|
||
Customer.friend_nickname,
|
||
Customer.friend_remark,
|
||
Customer.head_image_url,
|
||
sa.func.min(Message.timestamp).label("first_ts"),
|
||
)
|
||
.outerjoin(
|
||
Message,
|
||
sa.and_(
|
||
Message.channel == "wechat",
|
||
Message.owner_account_id == Customer.owner_account_id,
|
||
Message.talker_id == Customer.friend_id,
|
||
Message.timestamp.is_not(None),
|
||
),
|
||
)
|
||
.where(*base, Customer.add_time.is_(None))
|
||
.group_by(Customer.friend_id)
|
||
.having(sa.func.min(Message.timestamp) >= start)
|
||
.having(sa.func.min(Message.timestamp) <= end)
|
||
)
|
||
# 直接口径:add_time 本身在窗内。
|
||
direct = sa.select(
|
||
Customer.friend_id,
|
||
Customer.friend_nickname,
|
||
Customer.friend_remark,
|
||
Customer.head_image_url,
|
||
Customer.add_time,
|
||
).where(*base, Customer.add_time >= start, Customer.add_time <= end)
|
||
|
||
rows: list[dict[str, Any]] = [
|
||
{
|
||
"talker_id": fid,
|
||
"nickname": nick,
|
||
"avatar": head or None,
|
||
"remark": remark,
|
||
"time": add_time,
|
||
}
|
||
for fid, nick, remark, head, add_time in (
|
||
await session.execute(direct)
|
||
).all()
|
||
]
|
||
rows += [
|
||
{
|
||
"talker_id": fid,
|
||
"nickname": nick,
|
||
"avatar": head or None,
|
||
"remark": remark,
|
||
"time": first_ts,
|
||
}
|
||
for fid, nick, remark, head, first_ts in (
|
||
await session.execute(fallback)
|
||
).all()
|
||
]
|
||
rows.sort(key=lambda r: r["time"] or 0, reverse=True)
|
||
return len(rows), rows
|
||
|
||
|
||
async def _new_groups(
|
||
session: AsyncSession, wxid: str, start: int, end: int, detail: bool
|
||
) -> tuple[int, list[dict[str, Any]]]:
|
||
"""新群:chat_group.create_time 窗内(每员工一行,owner 过滤)。"""
|
||
conds = [
|
||
ChatGroup.channel == "wechat",
|
||
ChatGroup.owner_account_id == wxid,
|
||
ChatGroup.create_time >= start,
|
||
ChatGroup.create_time <= end,
|
||
]
|
||
count = (
|
||
await session.execute(
|
||
sa.select(sa.func.count()).select_from(ChatGroup).where(*conds)
|
||
)
|
||
).scalar_one()
|
||
if not detail:
|
||
return count, []
|
||
rows = (
|
||
await session.execute(
|
||
sa.select(ChatGroup.room_id, ChatGroup.room_name, ChatGroup.create_time)
|
||
.where(*conds)
|
||
.order_by(ChatGroup.create_time.desc(), ChatGroup.room_id.asc())
|
||
)
|
||
).all()
|
||
return count, [
|
||
{"room_id": rid, "room_name": name, "time": ts} for rid, name, ts in rows
|
||
]
|
||
|
||
|
||
async def _red_transfer(
|
||
session: AsyncSession, wxid: str, start: int, end: int, detail: bool
|
||
) -> tuple[int, list[dict[str, Any]]]:
|
||
"""红包+转账:wechat 消息 type IN (red_packet, transfer) 且窗内。"""
|
||
conds = [
|
||
Message.channel == "wechat",
|
||
Message.owner_account_id == wxid,
|
||
Message.type.in_(["red_packet", "transfer"]),
|
||
Message.timestamp >= start,
|
||
Message.timestamp <= end,
|
||
]
|
||
count = (
|
||
await session.execute(
|
||
sa.select(sa.func.count()).select_from(Message).where(*conds)
|
||
)
|
||
).scalar_one()
|
||
if not detail:
|
||
return count, []
|
||
msgs = (
|
||
await session.execute(
|
||
sa.select(
|
||
Message.msg_id,
|
||
Message.talker_id,
|
||
Message.room_id,
|
||
Message.type,
|
||
Message.text,
|
||
Message.timestamp,
|
||
Message.sender_name,
|
||
Message.raw_json,
|
||
)
|
||
.where(*conds)
|
||
.order_by(Message.timestamp.desc(), Message.msg_id.asc())
|
||
)
|
||
).all()
|
||
# nickname 优先 customer 昵称,回退 message raw_json.name(sender_name 即其
|
||
# 结构化列);群红包 talker 是发红包的成员,同口径查 customer。
|
||
talker_ids = [m.talker_id for m in msgs if m.talker_id]
|
||
nick_map: dict[str, str] = {}
|
||
if talker_ids:
|
||
nick_map = {
|
||
fid: nick
|
||
for fid, nick in (
|
||
await session.execute(
|
||
sa.select(Customer.friend_id, Customer.friend_nickname).where(
|
||
Customer.channel == "wechat",
|
||
Customer.owner_account_id == wxid,
|
||
Customer.friend_id.in_(talker_ids),
|
||
)
|
||
)
|
||
).all()
|
||
if nick
|
||
}
|
||
rows: list[dict[str, Any]] = []
|
||
for m in msgs:
|
||
raw_name = (m.raw_json or {}).get("name") if m.raw_json else None
|
||
rows.append(
|
||
{
|
||
"op_type": "红包" if m.type == "red_packet" else "转账",
|
||
"talker_id": m.talker_id,
|
||
"room_id": m.room_id or None,
|
||
"nickname": nick_map.get(m.talker_id or "") or m.sender_name or raw_name,
|
||
"amount": _extract_amount(m.text),
|
||
"time": m.timestamp,
|
||
"msg_id": m.msg_id,
|
||
}
|
||
)
|
||
return count, rows
|
||
|
||
|
||
async def _flag_customers(
|
||
session: AsyncSession,
|
||
wxid: str,
|
||
flag_cond: sa.ColumnElement,
|
||
*,
|
||
detail: bool,
|
||
) -> tuple[int, list[dict[str, Any]]]:
|
||
"""del_customers / blocked_customers 共用:现存量口径,忽略时间窗。
|
||
|
||
无时间字段→明细行 time 用 updated_at 毫秒(仅展示;update 触发即变)。
|
||
"""
|
||
conds = [Customer.channel == "wechat", Customer.owner_account_id == wxid, flag_cond]
|
||
count = (
|
||
await session.execute(
|
||
sa.select(sa.func.count()).select_from(Customer).where(*conds)
|
||
)
|
||
).scalar_one()
|
||
if not detail:
|
||
return count, []
|
||
rows = (
|
||
await session.execute(
|
||
sa.select(
|
||
Customer.friend_id,
|
||
Customer.friend_nickname,
|
||
Customer.friend_remark,
|
||
Customer.updated_at,
|
||
)
|
||
.where(*conds)
|
||
.order_by(Customer.updated_at.desc(), Customer.friend_id.asc())
|
||
)
|
||
).all()
|
||
return count, [
|
||
{
|
||
"talker_id": fid,
|
||
"nickname": nick or remark,
|
||
"time": _dt_ms(updated),
|
||
}
|
||
for fid, nick, remark, updated in rows
|
||
]
|
||
|
||
|
||
async def _del_customers(
|
||
session: AsyncSession, wxid: str, start: int, end: int, detail: bool
|
||
) -> tuple[int, list[dict[str, Any]]]:
|
||
"""删除客户:delete_flag=1 现存量(start/end 按口径忽略)。"""
|
||
return await _flag_customers(
|
||
session, wxid, Customer.delete_flag == 1, detail=detail
|
||
)
|
||
|
||
|
||
async def _blocked_customers(
|
||
session: AsyncSession, wxid: str, start: int, end: int, detail: bool
|
||
) -> tuple[int, list[dict[str, Any]]]:
|
||
"""流失(拉黑)客户:is_block=1 现存量(start/end 按口径忽略)。"""
|
||
return await _flag_customers(
|
||
session, wxid, Customer.is_block.is_(True), detail=detail
|
||
)
|
||
|
||
|
||
async def _voice_calls(
|
||
session: AsyncSession, wxid: str, start: int, end: int, detail: bool
|
||
) -> tuple[int, list[dict[str, Any]]]:
|
||
"""语音通话:先 wechat_account.user_id 换算(防直查 wxId 恒 0)再查。
|
||
|
||
明细字段语义对齐 CallList.vue / PhoneCallOut:
|
||
``connected`` 按 voice_call 极性 call_status==1(与前端 callStatusLabel
|
||
同判据:1=已接通,0/2=未接通);``media_url`` 本地 /api/media/{hash}
|
||
优先、OSS record_file 兜底(与 playRecord 同顺序)。账号无 user_id
|
||
(未同步/未知员工)→ 无法归属,返回空而非全量(防跨员工串数)。
|
||
"""
|
||
uid = (
|
||
await session.execute(
|
||
sa.select(WechatAccount.user_id).where(WechatAccount.wx_id == wxid)
|
||
)
|
||
).scalar_one_or_none()
|
||
if not uid:
|
||
return 0, []
|
||
conds = [
|
||
PhoneCall.channel == "voice_call",
|
||
PhoneCall.user_id == uid,
|
||
PhoneCall.start_call_time >= start,
|
||
PhoneCall.start_call_time <= end,
|
||
]
|
||
count = (
|
||
await session.execute(
|
||
sa.select(sa.func.count()).select_from(PhoneCall).where(*conds)
|
||
)
|
||
).scalar_one()
|
||
if not detail:
|
||
return count, []
|
||
calls = (
|
||
await session.execute(
|
||
sa.select(PhoneCall)
|
||
.where(*conds)
|
||
.order_by(PhoneCall.start_call_time.desc(), PhoneCall.call_id.asc())
|
||
)
|
||
).scalars().all()
|
||
rows = [
|
||
{
|
||
"id": c.call_id,
|
||
"peer": c.call_number,
|
||
"incoming": c.incoming_call,
|
||
"connected": c.call_status == 1,
|
||
"duration": c.call_duration,
|
||
"start_time": c.start_call_time,
|
||
"media_url": (
|
||
f"/api/media/{c.record_media_hash}" if c.record_media_hash else c.record_file
|
||
),
|
||
}
|
||
for c in calls
|
||
]
|
||
return count, rows
|
||
|
||
|
||
_AGGREGATORS = {
|
||
"new_customers": _new_customers,
|
||
"new_groups": _new_groups,
|
||
"red_transfer": _red_transfer,
|
||
"del_customers": _del_customers,
|
||
"blocked_customers": _blocked_customers,
|
||
"voice_calls": _voice_calls,
|
||
}
|
||
|
||
|
||
@router.get("/staff-stats")
|
||
async def get_staff_stats(
|
||
session: SessionDep,
|
||
wechat_id: Annotated[str, Query(min_length=1)],
|
||
start: Annotated[int | None, Query(ge=0)] = None,
|
||
end: Annotated[int | None, Query(ge=0)] = None,
|
||
kind: Annotated[KindParam | None, Query()] = None,
|
||
) -> StaffStatsCountsOut | StaffStatsDetailOut:
|
||
"""员工 7 类计数(无 kind)或某一类的明细列表(?kind=)。
|
||
|
||
窗口 [start, end] 均为 ms epoch 闭区间;缺省 = 近 7 天(云客统计默认窗)。
|
||
del_customers / blocked_customers 是现存量口径,窗口仅约束其余 4 类。
|
||
"""
|
||
if end is None:
|
||
end = int(time.time() * 1000)
|
||
if start is None:
|
||
start = end - 7 * _MS_PER_DAY
|
||
|
||
if kind is None:
|
||
counts: dict[str, int] = {}
|
||
for kind_key, aggregate in _AGGREGATORS.items():
|
||
counts[kind_key], _ = await aggregate(session, wechat_id, start, end, detail=False)
|
||
counts["sensitive_words"] = 0 # 占位,spec 决策
|
||
return StaffStatsCountsOut(**counts)
|
||
|
||
_, rows = await _AGGREGATORS[kind](session, wechat_id, start, end, detail=True)
|
||
return StaffStatsDetailOut(kind=kind, rows=rows)
|