diff --git a/yunke-server/app/db/models.py b/yunke-server/app/db/models.py index a76568d..9bce5b7 100644 --- a/yunke-server/app/db/models.py +++ b/yunke-server/app/db/models.py @@ -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", diff --git a/yunke-server/tests/conftest.py b/yunke-server/tests/conftest.py index eee01b3..9d1a299 100644 --- a/yunke-server/tests/conftest.py +++ b/yunke-server/tests/conftest.py @@ -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() diff --git a/yunke-server/tests/test_management_models.py b/yunke-server/tests/test_management_models.py new file mode 100644 index 0000000..2582ea9 --- /dev/null +++ b/yunke-server/tests/test_management_models.py @@ -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