feat(link): customer_account_link 账号↔客户多对多——共享好友员工维度对齐云客
用户报「Lily 好友数 36,点进员工客户页只有 12」。真机探明根因:那 24 个 是共享客户(同一人加多个员工),云客 per-employee 各算一个;而 customer 表 PK=friend_id 全局单行、归属 last-wins,其他员工名下必然「消失」。 抓取器本身每员工都拿全了(日志 declared=36 rows=36),丢在入库模型层。 - 新表 customer_account_link(渠道×账号×客户一行,per-relation 快照: 昵称/备注/头像/手机号/收发计数/最近沟通/add_time/fromType/isDel) - 写:repos.upsert_customer 随档案同事务维护(所有好友抓取路径一处挂全); 新增 prune_customer_links 对账 - 读:/api/customers?owner_account_id= 改读 link(与云客 contacts_num 口径 一致);该账号 link 为空时回退档案路径(过渡安全);与 dept/user 等 过滤组合时维持旧行为;CustomerOut 透传 from_type - 抓取器:friends/wa-friends 逐员工整批 upsert 后按名单对账(取满声明 total 才删,防半份名单误删) - 前端:StaffFriends 数据来源列优先 link 的 from_type 611 测试绿(link 新 6 例);本地验证空表回退/造数接管/from_type 透传。 上线后需重跑 friends+wa-friends 回填 link(存量关系无法从档案反推)。 Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
@@ -7,6 +7,12 @@ Read-only query over the mirrored ``customer`` table. Supports filtering by
|
||||
nickname / remark / friend_id / phone, ``%``/``_`` escaped as literals — the
|
||||
通讯录 search box). Results are ordered by ``last_chat_time desc`` (nulls
|
||||
last) then ``friend_id`` so the most-recently-active contacts surface first.
|
||||
|
||||
**owner_account_id 维度走 ``customer_account_link``**(2026-09-08):customer
|
||||
档案 PK=friend_id 全局单行、归属 last-wins,共享客户在其他员工名下会漏
|
||||
(实测 36 vs 12);link 表每个「渠道×账号×客户」一行,员工维度列表/计数与
|
||||
云客 per-employee 口径一致。带 department/user/add_time/last_chat 等其余
|
||||
过滤时回退档案路径(组合场景罕见,保持旧行为)。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -19,7 +25,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.api.deps import get_session
|
||||
from app.api.schemas import CustomerOut, Page
|
||||
from app.db.models import Customer, Staff, WechatAccount
|
||||
from app.db.models import Customer, CustomerAccountLink, Staff, WechatAccount
|
||||
|
||||
router = APIRouter(prefix="/api", tags=["customers"])
|
||||
|
||||
@@ -61,6 +67,63 @@ async def customers_declared_total(session: SessionDep) -> dict:
|
||||
}
|
||||
|
||||
|
||||
async def _list_owner_links(
|
||||
session: AsyncSession,
|
||||
*,
|
||||
page: int,
|
||||
page_size: int,
|
||||
channel: str | None,
|
||||
owner_account_id: str,
|
||||
friend_id: str | None,
|
||||
q: str | None,
|
||||
) -> Page[CustomerOut]:
|
||||
"""员工维度客户列表:读 ``customer_account_link``(渠道×账号×客户一行)。
|
||||
|
||||
排序/搜索语义与档案路径一致(last_chat_time desc、四列模糊);link 自带
|
||||
per-relation 快照(备注/收发计数按员工各一份,与云客 per-employee 列表
|
||||
相同)。CustomerOut 的档案专属列(raw_json/version/send_today 等)为 None。
|
||||
"""
|
||||
conditions = [CustomerAccountLink.owner_account_id == owner_account_id]
|
||||
if channel is not None:
|
||||
conditions.append(CustomerAccountLink.channel == channel)
|
||||
if friend_id is not None:
|
||||
conditions.append(CustomerAccountLink.friend_id == friend_id)
|
||||
if q:
|
||||
# LIKE 通配符转义(同档案路径):昵称/备注/微信号/手机号四列任一命中
|
||||
escaped = q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
|
||||
like = f"%{escaped}%"
|
||||
conditions.append(
|
||||
sa.or_(
|
||||
CustomerAccountLink.friend_nickname.like(like, escape="\\"),
|
||||
CustomerAccountLink.friend_remark.like(like, escape="\\"),
|
||||
CustomerAccountLink.friend_id.like(like, escape="\\"),
|
||||
CustomerAccountLink.phone.like(like, escape="\\"),
|
||||
)
|
||||
)
|
||||
|
||||
total = (
|
||||
await session.execute(
|
||||
sa.select(sa.func.count())
|
||||
.select_from(CustomerAccountLink)
|
||||
.where(*conditions)
|
||||
)
|
||||
).scalar_one()
|
||||
rows = (
|
||||
await session.execute(
|
||||
sa.select(CustomerAccountLink)
|
||||
.where(*conditions)
|
||||
.order_by(
|
||||
sa.nullslast(CustomerAccountLink.last_chat_time.desc()),
|
||||
CustomerAccountLink.friend_id.asc(),
|
||||
)
|
||||
.limit(page_size)
|
||||
.offset((page - 1) * page_size)
|
||||
)
|
||||
).scalars().all()
|
||||
items = [CustomerOut.model_validate(r) for r in rows]
|
||||
return Page(items=items, total=total, page=page, page_size=page_size)
|
||||
|
||||
|
||||
@router.get("/customers")
|
||||
async def list_customers(
|
||||
session: SessionDep,
|
||||
@@ -76,6 +139,34 @@ async def list_customers(
|
||||
q: Annotated[str | None, Query()] = None,
|
||||
) -> Page[CustomerOut]:
|
||||
"""List customers with optional filters and pagination."""
|
||||
# 员工维度优先走 link 表(共享好友每账号一行,见模块 docstring);该账号
|
||||
# link 还无数据时回退档案路径(过渡安全:回填前页面不至于空;也兼容
|
||||
# 只写档案的理论路径)。与其余仅档案路径支持的过滤组合时同样回退。
|
||||
if (
|
||||
owner_account_id is not None
|
||||
and department_id is None
|
||||
and user_id is None
|
||||
and add_time_since is None
|
||||
and last_chat_since is None
|
||||
):
|
||||
link_exists = (
|
||||
await session.execute(
|
||||
sa.select(sa.func.count())
|
||||
.select_from(CustomerAccountLink)
|
||||
.where(CustomerAccountLink.owner_account_id == owner_account_id)
|
||||
)
|
||||
).scalar_one()
|
||||
if link_exists:
|
||||
return await _list_owner_links(
|
||||
session,
|
||||
page=page,
|
||||
page_size=page_size,
|
||||
channel=channel,
|
||||
owner_account_id=owner_account_id,
|
||||
friend_id=friend_id,
|
||||
q=q,
|
||||
)
|
||||
|
||||
conditions = []
|
||||
if channel is not None:
|
||||
conditions.append(Customer.channel == channel)
|
||||
|
||||
@@ -43,6 +43,9 @@ class CustomerOut(BaseModel):
|
||||
# 客户页「发送条数/接收条数」列用,2026-09-08)
|
||||
send_num: int | None = None
|
||||
receive_num: int | None = None
|
||||
# 云客数据来源代码(customer_account_link 路径直读;档案路径为 None,
|
||||
# 前端回退 raw_json.fromType)
|
||||
from_type: int | None = None
|
||||
version: str | None = None
|
||||
# Verbatim upstream friend record (~60 camelCase fields). Surfaced so the
|
||||
# all-fields drawer can render every column the 云客 source page shows
|
||||
|
||||
@@ -262,6 +262,46 @@ class Customer(Base):
|
||||
)
|
||||
|
||||
|
||||
class CustomerAccountLink(Base):
|
||||
"""账号↔客户 多对多关联(customer_account_link,2026-09-08)。
|
||||
|
||||
``customer`` 表 PK=``friend_id`` 全局一客户一行,归属=最后抓取员工
|
||||
(last-wins)——共享客户(同一人加多个员工)在其他员工名下「消失」:
|
||||
实测 Lily 云客好友 36 vs 镜像页面 12,24 个全在别的员工名下。云客
|
||||
的口径是**每个账号各算一个**(per-employee 列表连备注/收发计数都按
|
||||
关系各存一份);本表每个「渠道×账号×客户」一行,由
|
||||
:func:`app.storage.repos.upsert_customer` 随档案同事务维护,员工维度
|
||||
的列表/计数(`/api/customers?owner_account_id=`)读这里。
|
||||
|
||||
快照列为员工客户页所需(昵称/备注/头像 per-relation);全字段档案
|
||||
(raw_json/60 列)仍在 ``customer`` 表按 friend_id 关联。
|
||||
"""
|
||||
|
||||
__tablename__ = "customer_account_link"
|
||||
|
||||
channel: Mapped[str] = mapped_column(String(16), primary_key=True)
|
||||
owner_account_id: Mapped[str] = mapped_column(String(128), primary_key=True)
|
||||
friend_id: Mapped[str] = mapped_column(String(128), primary_key=True)
|
||||
# per-relation 展示快照(云客好友记录按员工各一份;备注是员工自己的)
|
||||
friend_nickname: Mapped[str | None] = mapped_column(String(255), nullable=True)
|
||||
friend_remark: Mapped[str | None] = mapped_column(String(255), nullable=True)
|
||||
head_image_url: Mapped[str | None] = mapped_column(String(1024), nullable=True)
|
||||
real_name: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
||||
phone: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
||||
last_chat_time: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
|
||||
add_time: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
|
||||
send_num: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
receive_num: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
message_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
# 云客 fromType 代码(数据来源列;int,前端 fromTypeLabel 映射)
|
||||
from_type: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
# 同 customer.delete_flag 口径(int 0/1)
|
||||
delete_flag: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
|
||||
fetched_at: Mapped[dt.datetime] = _fetched_at()
|
||||
updated_at: Mapped[dt.datetime] = _updated_at()
|
||||
|
||||
|
||||
class ChatGroup(Base):
|
||||
"""聊天群 — source: ``POST /wechatweb/chatroom/getChatroomList``.
|
||||
|
||||
|
||||
@@ -63,6 +63,7 @@ from sqlalchemy.orm import DeclarativeBase
|
||||
from app.db.models import (
|
||||
ChatGroup,
|
||||
Customer,
|
||||
CustomerAccountLink,
|
||||
Department,
|
||||
GroupMember,
|
||||
Media,
|
||||
@@ -83,6 +84,7 @@ __all__ = [
|
||||
"upsert_chat_group",
|
||||
"upsert_customer",
|
||||
"upsert_customers",
|
||||
"prune_customer_links",
|
||||
"upsert_department",
|
||||
"upsert_media",
|
||||
"upsert_messages",
|
||||
@@ -380,9 +382,34 @@ async def upsert_customer(session: AsyncSession, rec: dict[str, Any]) -> None:
|
||||
"""Idempotently upsert one :class:`Customer` (PK = ``friend_id``).
|
||||
|
||||
Same ``friend_id`` twice -> one row, with non-PK fields refreshed from the
|
||||
later record.
|
||||
later record. 同时维护 ``customer_account_link``(渠道×账号×客户一行,
|
||||
2026-09-08):customer 档案全局单行 last-wins,共享客户的其他员工关系
|
||||
靠 link 表保留;员工维度列表/计数读 link(对齐云客 per-employee 口径)。
|
||||
"""
|
||||
await _upsert_one(session, Customer, rec)
|
||||
owner = rec.get("owner_account_id")
|
||||
if owner:
|
||||
raw = rec.get("raw_json") or {}
|
||||
link_rec = {
|
||||
"channel": rec.get("channel"),
|
||||
"owner_account_id": owner,
|
||||
"friend_id": rec.get("friend_id"),
|
||||
"friend_nickname": rec.get("friend_nickname"),
|
||||
"friend_remark": rec.get("friend_remark"),
|
||||
"head_image_url": rec.get("head_image_url"),
|
||||
"real_name": rec.get("real_name"),
|
||||
"phone": rec.get("phone"),
|
||||
"last_chat_time": rec.get("last_chat_time"),
|
||||
"add_time": rec.get("add_time"),
|
||||
"send_num": rec.get("send_num"),
|
||||
"receive_num": rec.get("receive_num"),
|
||||
"message_count": rec.get("message_count"),
|
||||
"delete_flag": rec.get("delete_flag"),
|
||||
}
|
||||
# fromType 只在 raw_json(云客数据来源代码,员工客户页列)
|
||||
if isinstance(raw, dict) and raw.get("fromType") is not None:
|
||||
link_rec["from_type"] = raw.get("fromType")
|
||||
await _upsert_one(session, CustomerAccountLink, link_rec)
|
||||
|
||||
|
||||
async def upsert_customers(session: AsyncSession, recs: list[dict[str, Any]]) -> int:
|
||||
@@ -392,10 +419,33 @@ async def upsert_customers(session: AsyncSession, recs: list[dict[str, Any]]) ->
|
||||
single row (the last write wins) at the DB layer.
|
||||
"""
|
||||
for rec in recs:
|
||||
await _upsert_one(session, Customer, rec)
|
||||
await upsert_customer(session, rec)
|
||||
return len(recs)
|
||||
|
||||
|
||||
async def prune_customer_links(
|
||||
session: AsyncSession,
|
||||
*,
|
||||
channel: str,
|
||||
owner_account_id: str,
|
||||
keep_friend_ids: list[str],
|
||||
) -> int:
|
||||
"""删除该账号名下**不在最新名单**里的 customer_account_link 行。
|
||||
|
||||
云客 isDel=0 只返回当前好友;抓取器逐员工整批 upsert 后调用本函数对账
|
||||
(离开名单=已不是好友,或好友转移)。**调用方必须保证名单完整**(翻满
|
||||
声明 total)再调——拿到半份名单就 prune 会误删。返回删除行数。
|
||||
"""
|
||||
conditions = [
|
||||
CustomerAccountLink.channel == channel,
|
||||
CustomerAccountLink.owner_account_id == owner_account_id,
|
||||
]
|
||||
if keep_friend_ids:
|
||||
conditions.append(CustomerAccountLink.friend_id.not_in(keep_friend_ids))
|
||||
result = await session.execute(sa.delete(CustomerAccountLink).where(*conditions))
|
||||
return int(result.rowcount or 0)
|
||||
|
||||
|
||||
async def upsert_messages(session: AsyncSession, recs: list[dict[str, Any]]) -> int:
|
||||
"""Batch idempotent upsert for :class:`Message` (PK = ``msg_id``).
|
||||
|
||||
|
||||
Reference in New Issue
Block a user