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

feat(management): 批次A六表模型——fact/宽表/通话小表+评分配置/快照/辅导

Browse Source 
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
2026-09-09 12:04:54 +08:00
parent 399ecc680f
commit e0c764997f
3 changed files with 356 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yunke-server/app/db/models.py
+232
View File
@@ -999,11 +999,243 @@ class SensitiveScanState(Base):
updated_at: Mapped[dt.datetime] = _updated_at()
# ==========================================================================
# 经营管理后台(2026-09-08 v2 spec,批次A)——fact/宽表/通话小表 + 三张业务表
# ==========================================================================
class FirstReplyFact(Base):
"""首响事实表(管理域)——一行 = 一个被回的 1:1 客户消息串(v2 §3.1)。
跨日精确串口径:按会话**全历史**计算,fr_ms = 员工首条回复 ts − 串末条
客户消息 ts,day = 串末条所在上海自然日。区别于 msg_daily_stats 首响
三列的日切近似(跨夜串整条丢失,v2 §1.1 #1)。只写 1:1(friend)会话
——群串语义不成立。由回填 CLI / 4h 滚动 / 夜间全量重算,**不挂写钩子**
(写钩子零改动)。用途:精确 P50/P90、2h/24h 回复率分子、下钻证据。
"""
__tablename__ = "first_reply_fact"
channel: Mapped[str] = mapped_column(String(16), primary_key=True)
owner_account_id: Mapped[str] = mapped_column(String(128), primary_key=True)
peer_id: Mapped[str] = mapped_column(String(128), primary_key=True)
thread_end_ts: Mapped[int] = mapped_column(BigInteger, primary_key=True)
reply_ts: Mapped[int] = mapped_column(BigInteger, primary_key=True)
# 串末条所在上海自然日 yyyymmdd(冗余物化,窗口/趋势索引)
day: Mapped[int] = mapped_column(Integer, nullable=False)
fr_ms: Mapped[int] = mapped_column(BigInteger, nullable=False)
__table_args__ = (Index("ix_fr_fact_day_channel", "day", "channel"),)
class EmployeeMetricDaily(Base):
"""员工×日管理指标宽表(v2 §3.2)——1:1 口径,管理域专用。
列来源(v2 §1.1 #2/#5):消息计数/负载列自 msg_daily_stats
(peer_kind='friend' 过滤 + account_map 归并);串类三列与首响日值由
first_reply_facts.compute_conversation 同源输出(串末条归属日)——
回复率分子分母同口径。metric_version 只影响派生列(生命周期),
默认 'v1'。account_map 映射不上或不在 staff 的 owner(幽灵员工)
过滤不落表(v2 §1.3 #11)。通话在 employee_call_daily(无渠道维)。
"""
__tablename__ = "employee_metric_daily"
day: Mapped[int] = mapped_column(Integer, primary_key=True)
channel: Mapped[str] = mapped_column(String(16), primary_key=True)
user_id: Mapped[str] = mapped_column(String(64), primary_key=True)
metric_version: Mapped[str] = mapped_column(
String(16), primary_key=True, default="v1", server_default="v1"
)
# 消息计数列(1:1 口径)
msg_sent: Mapped[int] = mapped_column(
Integer, nullable=False, default=0, server_default=sa.text("0")
)
msg_recv: Mapped[int] = mapped_column(
Integer, nullable=False, default=0, server_default=sa.text("0")
)
# 负载列(active_rooms 为群负载,唯一计入宽表的群维度)
active_friends: Mapped[int] = mapped_column(
Integer, nullable=False, default=0, server_default=sa.text("0")
)
active_rooms: Mapped[int] = mapped_column(
Integer, nullable=False, default=0, server_default=sa.text("0")
)
# 串类列(与 first_reply_fact 同源,按串末条归属日)
cust_threads: Mapped[int] = mapped_column(
Integer, nullable=False, default=0, server_default=sa.text("0")
)
threads_answered: Mapped[int] = mapped_column(
Integer, nullable=False, default=0, server_default=sa.text("0")
)
fr_sum_ms: Mapped[int] = mapped_column(
BigInteger, nullable=False, default=0, server_default=sa.text("0")
)
# 首响日值(趋势图用;评分用周期值从 fact 表窗口函数精确计算)
first_reply_p50_ms: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
first_reply_p90_ms: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
# 生命周期 4 列(距今天数口径,当日持有客户状态分布,当前归属)
new_customer_count: Mapped[int] = mapped_column(
Integer, nullable=False, default=0, server_default=sa.text("0")
)
cooling_customer_count: Mapped[int] = mapped_column(
Integer, nullable=False, default=0, server_default=sa.text("0")
)
dormant_customer_count: Mapped[int] = mapped_column(
Integer, nullable=False, default=0, server_default=sa.text("0")
)
reactivated_customer_count: Mapped[int] = mapped_column(
Integer, nullable=False, default=0, server_default=sa.text("0")
)
# fact 是否已覆盖当日(宽表生成前置条件;False=串类列可能滞后)
data_complete: Mapped[bool] = mapped_column(
Boolean, nullable=False, default=False, server_default=sa.text("0")
)
computed_at: Mapped[dt.datetime] = mapped_column(
sa.DateTime, nullable=False, server_default=sa.func.now()
)
class EmployeeCallDaily(Base):
"""员工×日通话小表(v2 §1.2 #9)——通话无消息渠道维度,独立于宽表。
phone_call.channel 是通话类型(phone_call/voice_call)非消息渠道,
塞进 (day,channel,user) 宽表会在 channel=all 聚合时双计。接通 =
call_duration > 0;followup = 通话后 24h 内该员工任一账号对该客户
有消息(Task 0 已核实:1:1 通话可匹配率 100%,@chatroom 群通话除外)。
"""
__tablename__ = "employee_call_daily"
day: Mapped[int] = mapped_column(Integer, primary_key=True)
user_id: Mapped[str] = mapped_column(String(64), primary_key=True)
call_count: Mapped[int] = mapped_column(
Integer, nullable=False, default=0, server_default=sa.text("0")
)
call_seconds: Mapped[int] = mapped_column(
BigInteger, nullable=False, default=0, server_default=sa.text("0")
)
answered_call_count: Mapped[int] = mapped_column(
Integer, nullable=False, default=0, server_default=sa.text("0")
)
call_followup_count: Mapped[int] = mapped_column(
Integer, nullable=False, default=0, server_default=sa.text("0")
)
computed_at: Mapped[dt.datetime] = mapped_column(
sa.DateTime, nullable=False, server_default=sa.func.now()
)
class EmployeeScoreConfig(Base):
"""评分配置版本(v1 §10.4/v2 §3.3)——config_json 存维度/指标/权重/
方向/阈值/样本门槛;激活前须过权重总和与阈值校验(批次B scoring 实现)。
MVP 无可视化编辑器,受控 CLI 种子维护。"""
__tablename__ = "employee_score_config"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
version: Mapped[str] = mapped_column(String(32), unique=True, nullable=False)
name: Mapped[str] = mapped_column(String(128), nullable=False)
# draft | active | retired
status: Mapped[str] = mapped_column(
String(16), nullable=False, default="draft", server_default="draft"
)
effective_from: Mapped[int | None] = mapped_column(Integer, nullable=True)
effective_to: Mapped[int | None] = mapped_column(Integer, nullable=True)
config_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=False)
created_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
approved_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
created_at: Mapped[dt.datetime] = mapped_column(
sa.DateTime, nullable=False, server_default=sa.func.now()
)
activated_at: Mapped[dt.datetime | None] = mapped_column(sa.DateTime, nullable=True)
class EmployeeScoreSnapshot(Base):
"""周/月考核快照(v1 §8/v2 §3.3)——不可变:冻结后业务层禁改评分字段;
重算走 supersedes_id 替代链。同批次×员工唯一。"""
__tablename__ = "employee_score_snapshot"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
snapshot_batch_id: Mapped[str] = mapped_column(String(64), nullable=False)
# week | month | custom
period_type: Mapped[str] = mapped_column(String(8), nullable=False)
period_start: Mapped[int] = mapped_column(Integer, nullable=False)
period_end: Mapped[int] = mapped_column(Integer, nullable=False)
user_id: Mapped[str] = mapped_column(String(64), nullable=False)
score_config_version: Mapped[str] = mapped_column(String(32), nullable=False)
metric_version: Mapped[str] = mapped_column(String(16), nullable=False)
total_score: Mapped[int | None] = mapped_column(Integer, nullable=True)
grade: Mapped[str | None] = mapped_column(String(2), nullable=True)
# scored | insufficient_sample | incomplete_data
score_status: Mapped[str] = mapped_column(
String(24), nullable=False, default="scored", server_default="scored"
)
dimension_scores_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
metric_details_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
data_as_of: Mapped[int] = mapped_column(Integer, nullable=False)
generated_at: Mapped[dt.datetime] = mapped_column(
sa.DateTime, nullable=False, server_default=sa.func.now()
)
frozen_at: Mapped[dt.datetime | None] = mapped_column(sa.DateTime, nullable=True)
frozen_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
supersedes_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
__table_args__ = (
sa.UniqueConstraint("snapshot_batch_id", "user_id",
name="uq_score_snapshot_batch_user"),
Index("ix_score_snapshot_period", "period_type", "period_start"),
)
class ManagementFollowup(Base):
"""管理辅导闭环(v1 §7.3/v2 §3.3)——来源指标→问题→管理动作→结果。"""
__tablename__ = "management_followup"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
employee_user_id: Mapped[str] = mapped_column(String(64), nullable=False)
# metric | anomaly | review | manual
source_type: Mapped[str] = mapped_column(String(16), nullable=False)
source_code: Mapped[str | None] = mapped_column(String(64), nullable=True)
source_period_start: Mapped[int | None] = mapped_column(Integer, nullable=True)
source_period_end: Mapped[int | None] = mapped_column(Integer, nullable=True)
source_value_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
reason: Mapped[str | None] = mapped_column(sa.Text, nullable=True)
action_type: Mapped[str | None] = mapped_column(String(32), nullable=True)
owner_manager_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
# open | in_progress | resolved | cancelled
status: Mapped[str] = mapped_column(
String(16), nullable=False, default="open", server_default="open"
)
due_at: Mapped[dt.datetime | None] = mapped_column(sa.DateTime, nullable=True)
result: Mapped[str | None] = mapped_column(sa.Text, nullable=True)
employee_feedback: Mapped[str | None] = mapped_column(sa.Text, nullable=True)
created_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
created_at: Mapped[dt.datetime] = mapped_column(
sa.DateTime, nullable=False, server_default=sa.func.now()
)
updated_at: Mapped[dt.datetime] = _updated_at()
closed_at: Mapped[dt.datetime | None] = mapped_column(sa.DateTime, nullable=True)
__table_args__ = (
Index("ix_followup_emp_status_due", "employee_user_id", "status", "due_at"),
Index("ix_followup_owner_status", "owner_manager_id", "status"),
)
__all__ = [
"ChatGroup",
"Customer",
"Department",
"EmployeeCallDaily",
"EmployeeMetricDaily",
"EmployeeScoreConfig",
"EmployeeScoreSnapshot",
"FirstReplyFact",
"GroupMember",
"ManagementFollowup",
"Media",
"Message",
"MomentsPost",
yunke-server/tests/conftest.py
+19
View File
@@ -112,3 +112,22 @@ async def make_client() -> AsyncIterator[
await client.aclose()
await engine.dispose()
app.dependency_overrides.clear()
@pytest.fixture
async def make_session() -> AsyncIterator[AsyncSession]:
"""独立内存 SQLite AsyncSession(不挂 API 依赖)——存储层/管理域测试用。
与 make_client 共用 StaticPool 单连接模式;每个测试独享全新 schema。
"""
engine = create_async_engine(
"sqlite+aiosqlite:///:memory:",
poolclass=StaticPool,
connect_args={"check_same_thread": False},
)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
async with SessionLocal() as s:
yield s
await engine.dispose()
yunke-server/tests/test_management_models.py
+105
View File
@@ -0,0 +1,105 @@
"""管理域 6 张新表的模型冒烟测试(批次A Task1)。
只验证:建表可用、主键/唯一约束生效、默认值正确。业务逻辑在
first_reply_facts / employee_metrics 的测试里。
"""
import datetime as dt # noqa: F401
import pytest
import sqlalchemy as sa
from sqlalchemy.exc import IntegrityError
from app.db.models import (
EmployeeCallDaily,
EmployeeMetricDaily,
EmployeeScoreConfig,
EmployeeScoreSnapshot,
FirstReplyFact,
ManagementFollowup,
)
async def test_first_reply_fact_roundtrip(make_session):
s = make_session
s.add(FirstReplyFact(channel="wechat", owner_account_id="wx1", peer_id="p1",
thread_end_ts=1_700_000_000_000,
reply_ts=1_700_000_600_000,
day=20260901, fr_ms=600_000))
await s.commit()
val = (await s.execute(sa.select(FirstReplyFact.fr_ms))).scalar_one()
assert val == 600_000
async def test_first_reply_fact_unique_sample(make_session):
s = make_session
kwargs = {"channel": "wechat", "owner_account_id": "wx1", "peer_id": "p1",
"thread_end_ts": 1, "reply_ts": 2, "day": 20260901, "fr_ms": 1}
s.add(FirstReplyFact(**kwargs))
await s.commit()
s.add(FirstReplyFact(**kwargs))
with pytest.raises(IntegrityError):
await s.commit()
async def test_metric_daily_versioning_coexists(make_session):
s = make_session
for v in ("v1", "v2"):
s.add(EmployeeMetricDaily(day=20260901, channel="wechat",
user_id="u1", metric_version=v))
await s.commit()
n = (await s.execute(
sa.select(sa.func.count()).select_from(EmployeeMetricDaily))).scalar_one()
assert n == 2
async def test_metric_daily_defaults_zero_not_none(make_session):
s = make_session
s.add(EmployeeMetricDaily(day=20260901, channel="wechat", user_id="u1"))
await s.commit()
row = (await s.execute(sa.select(EmployeeMetricDaily))).scalar_one()
assert row.msg_sent == 0 and row.cust_threads == 0 and row.fr_sum_ms == 0
assert row.metric_version == "v1"
assert row.data_complete is False or row.data_complete == 0 # SQLite 0/1
async def test_call_daily_roundtrip(make_session):
s = make_session
s.add(EmployeeCallDaily(day=20260901, user_id="u1", call_count=2,
call_seconds=164, answered_call_count=1,
call_followup_count=1))
await s.commit()
row = (await s.execute(sa.select(EmployeeCallDaily))).scalar_one()
assert row.call_seconds == 164
async def test_score_config_version_unique(make_session):
s = make_session
s.add(EmployeeScoreConfig(version="v1.0", name="一期试算",
config_json={"dimensions": {}}))
await s.commit()
s.add(EmployeeScoreConfig(version="v1.0", name="重复",
config_json={}))
with pytest.raises(IntegrityError):
await s.commit()
async def test_snapshot_batch_user_unique(make_session):
s = make_session
common = {"snapshot_batch_id": "b1", "period_type": "week",
"period_start": 20260901, "period_end": 20260907,
"score_config_version": "v1.0", "metric_version": "v1",
"data_as_of": 20260907}
s.add(EmployeeScoreSnapshot(user_id="u1", **common))
await s.commit()
s.add(EmployeeScoreSnapshot(user_id="u1", **common))
with pytest.raises(IntegrityError):
await s.commit()
async def test_followup_defaults(make_session):
s = make_session
s.add(ManagementFollowup(employee_user_id="u1", source_type="manual"))
await s.commit()
row = (await s.execute(sa.select(ManagementFollowup))).scalar_one()
assert row.status == "open"
assert row.created_at is not None