xinyouhe/yunke
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(api): staff-stats评审修复(GROUP BY显式列防ONLY_FULL_GROUP_BY/金额千分位/detail死参数/2边缘测试)

Browse Source 
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
浅念
2026-08-25 16:36:07 +08:00
parent b95fc5fa6f
commit 4baa8639ba
2 changed files with 114 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yunke-server/app/api/routers/staff_stats.py
+33 -10
View File
@@ -61,9 +61,10 @@ KindParam = Literal[
# 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.]+)")
# 金额可带千分位逗号(``￥1,000.00``)→ 捕获组为 [\d.,]+,float() 前剔逗号。
_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:
@@ -78,8 +79,8 @@ def _extract_amount(text: str | None) -> float | None:
if not m:
return None
try:
return float(m.group(1))
except ValueError: # pragma: no cover — 正则已限定 [\d.]+,防御畸形串
return float(m.group(1).replace(",", ""))
except ValueError: # pragma: no cover — 正则已限定 [\d.,]+,防御畸形串
return None
@@ -91,9 +92,13 @@ def _dt_ms(value: Any) -> int | None:
async def _new_customers(
session: AsyncSession, wxid: str, start: int, end: int, detail: bool
session: AsyncSession, wxid: str, start: int, end: int
) -> tuple[int, list[dict[str, Any]]]:
"""新客户:add_time 窗内;add_time IS NULL 用首条消息时间兜底。"""
"""新客户:add_time 窗内;add_time IS NULL 用首条消息时间兜底。
无 ``detail`` 分支——计数模式有意复用明细物化(窗内新增客户通常
几十行,``len()`` 即可;COUNT 需 direct+fallback 双查询,不值)。
"""
base = [Customer.channel == "wechat", Customer.owner_account_id == wxid]
# 兜底口径:每个 add_time 为空的客户,取其与该员工首条消息 min(timestamp)
# (LEFT JOIN 走 ix_message_channel_owner_talker_time,HAVING 在 SQL 侧过滤窗)。
@@ -115,7 +120,15 @@ async def _new_customers(
),
)
.where(*base, Customer.add_time.is_(None))
.group_by(Customer.friend_id)
# GROUP BY 显式列出全部裸选列:friend_id 是 Customer 唯一 PK,其余列
# 函数依赖于它→分组结果不变;MySQL 8 FD 推断可容忍裸列,但显式列表
# 在 ONLY_FULL_GROUP_BY 下部署安全(SQLite 同样兼容)。
.group_by(
Customer.friend_id,
Customer.friend_nickname,
Customer.friend_remark,
Customer.head_image_url,
)
.having(sa.func.min(Message.timestamp) >= start)
.having(sa.func.min(Message.timestamp) <= end)
)
@@ -399,9 +412,19 @@ async def get_staff_stats(
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)
if kind_key == "new_customers": # 无 detail 参数(见其 docstring)
counts[kind_key], _ = await aggregate(session, wechat_id, start, end)
else:
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)
if kind == "new_customers": # 同上:无 detail 参数
_, rows = await _AGGREGATORS[kind](session, wechat_id, start, end)
else:
_, rows = await _AGGREGATORS[kind](
session, wechat_id, start, end, detail=True
)
return StaffStatsDetailOut(kind=kind, rows=rows)