浅念
133 lines
5.3 KiB
Python
133 lines
5.3 KiB
Python
"""employee_call_daily / employee_metric_daily 维护层(管理域,批次A)。
|
|
|
|
数据流(v2 §4):
|
|
msg_daily_stats(friend) + first_reply_facts 同源扫描 + customer 生命周期
|
|
→ employee_metric_daily (day, channel, user, metric_version) [Task 6]
|
|
phone_call(按 user_id+start_call_time 日聚合)
|
|
→ employee_call_daily (day, user)——通话无渠道维(v2 §1.2 #9) [本任务]
|
|
|
|
幽灵员工过滤(v2 §1.3 #11):owner 映射不上或不在 staff 表的不落表。
|
|
重算语义:按 day delete+insert,幂等;夜间全量 + (批次F)4h 滚动。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from collections.abc import Iterable
|
|
|
|
import sqlalchemy as sa
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.db.models import EmployeeCallDaily, Message, PhoneCall, Staff
|
|
from app.storage.staff_summary import account_map
|
|
from app.storage.summary import day_bounds_ms
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# 通话类型 → 消息渠道(Task0 核实:channel 列可判别,voice_call=微信/
|
|
# wa_voice_call=WA;followup 消息查询按此收窄)
|
|
_CALL_CHANNEL_TO_MSG: dict[str, str] = {"voice_call": "wechat", "wa_voice_call": "whatsapp"}
|
|
|
|
|
|
async def known_user_ids(session: AsyncSession) -> set[str]:
|
|
"""员工主数据 user_id 集合(staff 表;Task0 已核实 41 行全在职)。"""
|
|
rows = (await session.execute(sa.select(Staff.user_id))).scalars().all()
|
|
return {u for u in rows if u}
|
|
|
|
|
|
async def _followup_exists(
|
|
session: AsyncSession,
|
|
msg_channel: str,
|
|
owner_accounts: list[str],
|
|
peer: str,
|
|
call_ts_ms: int,
|
|
window_ms: int = 86_400_000,
|
|
) -> bool:
|
|
"""通话后 24h 内员工任一账号对该客户有 1:1 消息。
|
|
|
|
按通话类型收窄消息渠道(Task0:voice_call→wechat/wa_voice_call→whatsapp,
|
|
1:1 通话可匹配率 100%,@chatroom 群通话天然不匹配 1:1 消息)。"""
|
|
if not owner_accounts or not peer:
|
|
return False
|
|
n = (
|
|
await session.execute(
|
|
sa.select(sa.func.count()).select_from(Message).where(
|
|
Message.channel == msg_channel,
|
|
Message.owner_account_id.in_(owner_accounts),
|
|
Message.talker_id == peer,
|
|
sa.or_(Message.room_id.is_(None), Message.room_id == ""),
|
|
Message.timestamp.is_not(None),
|
|
Message.timestamp > call_ts_ms,
|
|
Message.timestamp <= call_ts_ms + window_ms,
|
|
)
|
|
)
|
|
).scalar_one()
|
|
return bool(n)
|
|
|
|
|
|
async def recompute_call_days(session: AsyncSession, days: Iterable[int]) -> int:
|
|
"""重算给定日的通话小表:按 day delete+insert,幂等。返回写入行数。
|
|
|
|
口径:接通 = call_duration > 0(Task0:与 call_status=1 差 31 条/1.0%,
|
|
取 dur>0 与现有 roadmap 口径一致);call_seconds 单位=秒(Task0 已核实);
|
|
followup 只对已接通通话判定(未接通的回拨语义留二期)。
|
|
"""
|
|
# 员工→该渠道账号(followup 判定用,按消息渠道分组)
|
|
acct = {ch: await account_map(session, ch) for ch in ("wechat", "whatsapp")}
|
|
rev: dict[str, dict[str, list[str]]] = {} # uid → msg_channel → owners
|
|
for msg_ch, amap in acct.items():
|
|
for owner, uid in amap.items():
|
|
rev.setdefault(uid, {}).setdefault(msg_ch, []).append(owner)
|
|
|
|
written = 0
|
|
staff_ids = await known_user_ids(session) # 幽灵过滤(v2 §1.3 #11)
|
|
for day in sorted(set(days)):
|
|
lo, hi = day_bounds_ms(day)
|
|
calls = (
|
|
await session.execute(
|
|
sa.select(
|
|
PhoneCall.user_id,
|
|
PhoneCall.channel,
|
|
PhoneCall.call_number,
|
|
PhoneCall.start_call_time,
|
|
PhoneCall.call_duration,
|
|
).where(
|
|
PhoneCall.start_call_time.is_not(None),
|
|
PhoneCall.start_call_time >= lo,
|
|
PhoneCall.start_call_time < hi,
|
|
)
|
|
)
|
|
).all()
|
|
await session.execute(
|
|
sa.delete(EmployeeCallDaily).where(EmployeeCallDaily.day == day)
|
|
)
|
|
# (uid, call_type, number, ts, dur)——msg_channel 按通话类型映射;
|
|
# 未知通话类型(如蜂窝 phone_call)无消息渠道可查,followup 判
|
|
# False 但仍计数
|
|
agg: dict[str, dict] = {}
|
|
for uid, call_ch, number, ts, dur in calls:
|
|
if not uid or uid not in staff_ids:
|
|
continue
|
|
a = agg.setdefault(uid, {"cnt": 0, "sec": 0, "answered": 0, "followup": 0})
|
|
a["cnt"] += 1
|
|
a["sec"] += int(dur or 0)
|
|
if (dur or 0) > 0:
|
|
a["answered"] += 1
|
|
msg_ch = _CALL_CHANNEL_TO_MSG.get(call_ch or "")
|
|
if msg_ch and number:
|
|
owners = rev.get(uid, {}).get(msg_ch, [])
|
|
if await _followup_exists(session, msg_ch, owners, str(number), int(ts)):
|
|
a["followup"] += 1
|
|
for uid, a in sorted(agg.items()):
|
|
session.add(
|
|
EmployeeCallDaily(
|
|
day=day,
|
|
user_id=uid,
|
|
call_count=a["cnt"],
|
|
call_seconds=a["sec"],
|
|
answered_call_count=a["answered"],
|
|
call_followup_count=a["followup"],
|
|
)
|
|
)
|
|
written += 1
|
|
return written
|