feat(group-stats): 微信群列表对齐云客「微信群统计」——getChatroomList 数据源+窗内统计
用户报微信群列表数据/内容不对,应对齐云客「沟通→微信号管理→微信群统计」 (/cms/wechat/wechatgroup)。真机探明(表单登录+页面导航抓 network): - 数据源=POST /wechatweb/chatroom/getChatroomList(form:departmentId+ startTime/endTime 窗+分页;信封 data.overviews[]/totalCount 1,854 行) - 云客窗是任意时间窗实时算:沟通消息数实测同群全史 417/7天 36/单日 6 - **修正存量映射错误**:member_count 原存 friendNum(员工好友数),真群成员 数是 friendCount/getChatroomList.memberCount - 新增/退出成员数云客按窗实时算,本地无事件流 → 按天积累 实现: - ChatGroup 补 group_remark 列(备注仅 friend_list/es 携带,两归一化均 「有值才写」防互清);迁移含存量 raw_json 回填(须先于基线抓取执行) - 新表 chat_group_member_stats(渠道×账号×群×天一行 added/lost) - 新实体 chatroom-stats:全窗刷档案列(真成员数/群备注/员工信息,owner 经 wechat_account alias 解析)+ --day 单日窗落进出日行;full_sync 5c 步 - /api/groups 扩展:msg_start/msg_end 窗内聚合(window_msg_count= msg_daily_stats 按天和、window_added/lost=日表和)、last_start/end 最新消息窗、group_remark 子串;行补员工微信/员工信息关联列 - GroupList 微信 tab:统计时段(默认近7天)/最新消息时间段/群名称/群备注/ 员工下拉/查询 + 9 列(群信息|群备注|员工微信|员工信息|群成员数量| 新增成员数|退出成员数|沟通消息数|操作);WA tab 不动 700 测试绿(新增窗聚合/筛选/员工列 3 例)。本地实测:基线 1,854 群=云客 total 一字不差,今日日窗 added 407/lost 166 落表,页面 9 列/员工列渲染正常。 Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
@@ -5,6 +5,10 @@
|
||||
``owner_user_id`` column) and ``q`` (substring match on ``room_name``,
|
||||
``%``/``_`` escaped as literals — the 通讯录群 tab search box), ordered by
|
||||
``last_chat_time desc`` (nulls last) so recently-active groups surface first.
|
||||
微信群统计扩展(2026-09-09,云客「微信群统计」同款):``msg_start``/``msg_end``
|
||||
统计时段(窗内输出沟通消息数/新增/退出成员数聚合)、``last_start``/
|
||||
``last_end`` 最新消息时间段、``group_remark`` 群备注子串;行补员工微信/
|
||||
员工信息列(wechat_account/staff 关联)。
|
||||
* ``GET /api/groups/{room_id}/members`` -> ``list[GroupMemberOut]`` — the
|
||||
members of a single group (small set, no pagination needed).
|
||||
"""
|
||||
@@ -19,7 +23,15 @@ from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.api.deps import get_session
|
||||
from app.api.schemas import GroupMemberOut, GroupOut, Page
|
||||
from app.db.models import ChatGroup, GroupMember
|
||||
from app.db.models import (
|
||||
ChatGroup,
|
||||
ChatGroupMemberStats,
|
||||
GroupMember,
|
||||
MsgDailyStats,
|
||||
Staff,
|
||||
WechatAccount,
|
||||
)
|
||||
from app.storage.summary import day_key
|
||||
|
||||
router = APIRouter(prefix="/api", tags=["groups"])
|
||||
|
||||
@@ -38,6 +50,11 @@ async def list_groups(
|
||||
user_id: Annotated[str | None, Query()] = None,
|
||||
room_id: Annotated[str | None, Query()] = None,
|
||||
q: Annotated[str | None, Query()] = None,
|
||||
group_remark: Annotated[str | None, Query()] = None,
|
||||
last_start: Annotated[int | None, Query(ge=0)] = None,
|
||||
last_end: Annotated[int | None, Query(ge=0)] = None,
|
||||
msg_start: Annotated[int | None, Query(ge=0)] = None,
|
||||
msg_end: Annotated[int | None, Query(ge=0)] = None,
|
||||
) -> Page[GroupOut]:
|
||||
"""List chat groups with optional ``channel`` / ``owner_account_id`` /
|
||||
``user_id`` / ``room_id`` / ``q`` filters.
|
||||
@@ -47,6 +64,13 @@ async def list_groups(
|
||||
filter on ``GET /api/customers``. ``room_id`` is an exact-match lookup
|
||||
(composite PK: the same room may return one row per owner; chat page
|
||||
group-name resolution). ``q`` is a substring match on ``room_name``.
|
||||
|
||||
微信群统计参数(2026-09-09):``last_start``/``last_end`` 过滤
|
||||
``last_chat_time`` 范围(最新消息时间段);``msg_start``/``msg_end``
|
||||
统计时段——窗内聚合三列(``window_msg_count`` 沟通消息数=
|
||||
msg_daily_stats 按天求和、``window_added``/``window_lost`` 成员进出=
|
||||
chat_group_member_stats 按天求和,上线日起有数据);``group_remark``
|
||||
群备注子串。员工微信/员工信息列经 wechat_account/staff 关联补齐。
|
||||
"""
|
||||
conditions = []
|
||||
if channel is not None:
|
||||
@@ -57,10 +81,17 @@ async def list_groups(
|
||||
# LIKE 通配符转义: %/_ 按字面匹配(镜像 customers/messages 的 q 处理)
|
||||
escaped = q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
|
||||
conditions.append(ChatGroup.room_name.like(f"%{escaped}%", escape="\\"))
|
||||
if group_remark:
|
||||
escaped_r = group_remark.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
|
||||
conditions.append(ChatGroup.group_remark.like(f"%{escaped_r}%", escape="\\"))
|
||||
if room_id is not None:
|
||||
conditions.append(ChatGroup.room_id == room_id)
|
||||
if user_id is not None:
|
||||
conditions.append(ChatGroup.owner_user_id == user_id)
|
||||
if last_start is not None:
|
||||
conditions.append(ChatGroup.last_chat_time >= last_start)
|
||||
if last_end is not None:
|
||||
conditions.append(ChatGroup.last_chat_time <= last_end)
|
||||
|
||||
total_stmt = sa.select(sa.func.count()).select_from(ChatGroup)
|
||||
if conditions:
|
||||
@@ -79,7 +110,84 @@ async def list_groups(
|
||||
.offset((page - 1) * page_size)
|
||||
)
|
||||
rows = (await session.execute(list_stmt)).scalars().all()
|
||||
items = [GroupOut.model_validate(row) for row in rows]
|
||||
|
||||
# ---- 员工微信/员工信息列(wechat_account/staff 关联,本页行集) ----
|
||||
acc_map: dict[str, tuple[str | None, str | None, str | None]] = {}
|
||||
owners = {r.owner_account_id for r in rows if r.owner_account_id}
|
||||
if owners:
|
||||
for wx, nick, alias, raw in (await session.execute(
|
||||
sa.select(
|
||||
WechatAccount.wx_id, WechatAccount.nick_name,
|
||||
WechatAccount.alias, WechatAccount.raw_json,
|
||||
).where(WechatAccount.wx_id.in_(owners))
|
||||
)).all():
|
||||
avatar = None
|
||||
if isinstance(raw, dict):
|
||||
h = raw.get("headImageUrl")
|
||||
avatar = h if isinstance(h, str) else None
|
||||
acc_map[wx] = (nick, alias, avatar)
|
||||
staff_map: dict[str, tuple[str | None, str | None]] = {}
|
||||
uids = {r.owner_user_id for r in rows if r.owner_user_id}
|
||||
if uids:
|
||||
for uid, name, phone in (await session.execute(
|
||||
sa.select(Staff.user_id, Staff.name, Staff.phone)
|
||||
.where(Staff.user_id.in_(uids))
|
||||
)).all():
|
||||
staff_map[uid] = (name, phone)
|
||||
|
||||
# ---- 统计时段窗内聚合(msg_daily_stats 沟通消息数 + member_stats 进出) ----
|
||||
msg_cnt: dict[tuple[str, str], int] = {}
|
||||
added_map: dict[tuple[str, str], int] = {}
|
||||
lost_map: dict[tuple[str, str], int] = {}
|
||||
if msg_start is not None and msg_end is not None and rows:
|
||||
d0, d1 = day_key(msg_start), day_key(msg_end)
|
||||
pairs = [(r.owner_account_id, r.room_id) for r in rows]
|
||||
owners_in = {p[0] for p in pairs}
|
||||
rooms_in = {p[1] for p in pairs}
|
||||
for owner, room, cnt in (await session.execute(
|
||||
sa.select(
|
||||
MsgDailyStats.owner_account_id, MsgDailyStats.peer_id,
|
||||
sa.func.coalesce(sa.func.sum(
|
||||
MsgDailyStats.send_cnt + MsgDailyStats.recv_cnt), 0),
|
||||
).where(
|
||||
MsgDailyStats.channel == (channel or "wechat"),
|
||||
MsgDailyStats.peer_kind == "room",
|
||||
MsgDailyStats.owner_account_id.in_(owners_in),
|
||||
MsgDailyStats.peer_id.in_(rooms_in),
|
||||
MsgDailyStats.day >= d0, MsgDailyStats.day <= d1,
|
||||
).group_by(MsgDailyStats.owner_account_id, MsgDailyStats.peer_id)
|
||||
)).all():
|
||||
msg_cnt[(owner, room)] = int(cnt)
|
||||
for owner, room, a, l in (await session.execute(
|
||||
sa.select(
|
||||
ChatGroupMemberStats.owner_account_id, ChatGroupMemberStats.room_id,
|
||||
sa.func.coalesce(sa.func.sum(ChatGroupMemberStats.added), 0),
|
||||
sa.func.coalesce(sa.func.sum(ChatGroupMemberStats.lost), 0),
|
||||
).where(
|
||||
ChatGroupMemberStats.channel == (channel or "wechat"),
|
||||
ChatGroupMemberStats.owner_account_id.in_(owners_in),
|
||||
ChatGroupMemberStats.room_id.in_(rooms_in),
|
||||
ChatGroupMemberStats.day >= d0, ChatGroupMemberStats.day <= d1,
|
||||
).group_by(ChatGroupMemberStats.owner_account_id, ChatGroupMemberStats.room_id)
|
||||
)).all():
|
||||
added_map[(owner, room)] = int(a)
|
||||
lost_map[(owner, room)] = int(l)
|
||||
|
||||
items = []
|
||||
for r in rows:
|
||||
out = GroupOut.model_validate(r)
|
||||
acc = acc_map.get(r.owner_account_id or "")
|
||||
if acc:
|
||||
out.staff_nickname, out.staff_wx_alias, out.staff_avatar = acc
|
||||
st = staff_map.get(r.owner_user_id or "")
|
||||
if st:
|
||||
out.staff_name, out.staff_phone = st
|
||||
if msg_start is not None and msg_end is not None:
|
||||
key = (r.owner_account_id or "", r.room_id)
|
||||
out.window_msg_count = msg_cnt.get(key, 0)
|
||||
out.window_added = added_map.get(key, 0)
|
||||
out.window_lost = lost_map.get(key, 0)
|
||||
items.append(out)
|
||||
return Page(items=items, total=total, page=page, page_size=page_size)
|
||||
|
||||
|
||||
|
||||
@@ -199,6 +199,17 @@ class GroupOut(BaseModel):
|
||||
owner_avatar: str | None = None
|
||||
friend_count: int | None = None
|
||||
create_time: int | None = None # 建群时间 ms epoch
|
||||
# 微信群统计页(2026-09-09,云客 getChatroomList 同款)
|
||||
group_remark: str | None = None # 群备注
|
||||
staff_nickname: str | None = None # 员工微信·昵称(wechat_account.nick_name)
|
||||
staff_wx_alias: str | None = None # 员工微信·微信号(wechat_account.alias)
|
||||
staff_avatar: str | None = None # 员工微信·头像
|
||||
staff_name: str | None = None # 员工信息·姓名(staff.name)
|
||||
staff_phone: str | None = None # 员工信息·手机(staff.phone)
|
||||
# 统计时段窗内聚合(仅请求带 msg_start/msg_end 时非 None):
|
||||
window_msg_count: int | None = None # 沟通消息数(窗内,msg_daily_stats)
|
||||
window_added: int | None = None # 新增成员数(窗内,按天表求和)
|
||||
window_lost: int | None = None # 退出成员数(窗内)
|
||||
raw_json: dict[str, Any] | None = None
|
||||
|
||||
|
||||
|
||||
@@ -344,6 +344,9 @@ class ChatGroup(Base):
|
||||
owner_avatar: Mapped[str | None] = mapped_column(String(1024), nullable=True)
|
||||
friend_count: Mapped[int | None] = mapped_column(Integer, nullable=True) # 群成员中员工好友数
|
||||
create_time: Mapped[int | None] = mapped_column(BigInteger, nullable=True) # 建群时间 ms epoch
|
||||
# 群备注(微信群统计页「群备注」列;getChatroomList.remark /
|
||||
# friend_list/es 的 friendWxRemark,2026-09-09)
|
||||
group_remark: Mapped[str | None] = mapped_column(String(255), nullable=True)
|
||||
|
||||
version: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
||||
raw_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
|
||||
@@ -362,6 +365,34 @@ class ChatGroup(Base):
|
||||
)
|
||||
|
||||
|
||||
class ChatGroupMemberStats(Base):
|
||||
"""群成员进出按天统计(chat_group_member_stats,2026-09-09)。
|
||||
|
||||
云客「微信群统计」的新增/退出成员数是 ``getChatroomList`` 按
|
||||
``startTime~endTime`` 任意窗实时算的;本地无成员进出事件流,改为每夜
|
||||
抓一次「当日窗」的 addMembers/lossMembers 落一行,页面任意日窗=按天
|
||||
求和。**上线日起积累**(历史日期无数据;chat_group.add_members/
|
||||
loss_members 旧列保留作最近一次抓取窗的快照)。
|
||||
"""
|
||||
|
||||
__tablename__ = "chat_group_member_stats"
|
||||
|
||||
channel: Mapped[str] = mapped_column(String(16), primary_key=True)
|
||||
owner_account_id: Mapped[str] = mapped_column(String(128), primary_key=True)
|
||||
room_id: Mapped[str] = mapped_column(String(128), primary_key=True)
|
||||
# yyyymmdd(上海自然日,同 msg_daily_stats 口径)
|
||||
day: Mapped[int] = mapped_column(Integer, primary_key=True)
|
||||
added: Mapped[int] = mapped_column(
|
||||
Integer, nullable=False, default=0, server_default=sa.text("0")
|
||||
)
|
||||
lost: Mapped[int] = mapped_column(
|
||||
Integer, nullable=False, default=0, server_default=sa.text("0")
|
||||
)
|
||||
|
||||
fetched_at: Mapped[dt.datetime] = _fetched_at()
|
||||
updated_at: Mapped[dt.datetime] = _updated_at()
|
||||
|
||||
|
||||
class GroupMember(Base):
|
||||
"""群成员 — composite PK (channel, room_id, member_id).
|
||||
|
||||
|
||||
@@ -901,7 +901,7 @@ def _normalize_chat_group(record: dict[str, Any]) -> dict[str, Any]:
|
||||
``'YYYY-MM-DD HH:mm:ss'`` 字符串,另兼容 ms-int/数字串,统一归一为
|
||||
ms-epoch int,复用 ``_parse_add_time`` 的双形态解析)
|
||||
"""
|
||||
return {
|
||||
rec = {
|
||||
"room_id": record.get("friendWxId"),
|
||||
"channel": "wechat",
|
||||
"room_name": record.get("friendWxNickname"),
|
||||
@@ -929,6 +929,12 @@ def _normalize_chat_group(record: dict[str, Any]) -> dict[str, Any]:
|
||||
"version": record.get("version"),
|
||||
"raw_json": record,
|
||||
}
|
||||
# 群备注(2026-09-09 微信群统计):friend_list/es 是唯一备注源
|
||||
# (getChatroomList 行无备注键);**有值才写**——空值不放键,upsert 不
|
||||
# 更新该列,不会被无备注的抓取轮次清空
|
||||
if record.get("friendWxRemark"):
|
||||
rec["group_remark"] = record["friendWxRemark"]
|
||||
return rec
|
||||
|
||||
|
||||
async def scrape_wechat_groups(
|
||||
|
||||
@@ -62,6 +62,7 @@ from sqlalchemy.orm import DeclarativeBase
|
||||
|
||||
from app.db.models import (
|
||||
ChatGroup,
|
||||
ChatGroupMemberStats,
|
||||
Customer,
|
||||
CustomerAccountLink,
|
||||
Department,
|
||||
@@ -82,6 +83,7 @@ _logger = logging.getLogger(__name__)
|
||||
|
||||
__all__ = [
|
||||
"upsert_chat_group",
|
||||
"upsert_chat_group_member_stats",
|
||||
"upsert_customer",
|
||||
"upsert_customers",
|
||||
"prune_customer_links",
|
||||
@@ -631,6 +633,18 @@ async def upsert_chat_group(session: AsyncSession, rec: dict[str, Any]) -> None:
|
||||
await _upsert_one(session, GroupMember, member_rec)
|
||||
|
||||
|
||||
async def upsert_chat_group_member_stats(
|
||||
session: AsyncSession, rec: dict[str, Any]
|
||||
) -> None:
|
||||
"""Idempotently upsert one :class:`ChatGroupMemberStats` day row.
|
||||
|
||||
PK = (channel, owner_account_id, room_id, day)。每夜「当日窗」抓取的
|
||||
addMembers/lossMembers 落一行;同日重跑刷新(幂等)。页面任意日窗的
|
||||
新增/退出 = 按天求和(2026-09-09 微信群统计对齐)。
|
||||
"""
|
||||
await _upsert_one(session, ChatGroupMemberStats, rec)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# PhoneCall (voice_call) upsert (M2, Task 2.4)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -0,0 +1,80 @@
|
||||
"""chat_group 补 group_remark 列 + chat_group_member_stats 按天进出表
|
||||
|
||||
Revision ID: c9d0e1f2a3b4
|
||||
Revises: b7c8d9e0f1a2
|
||||
Create Date: 2026-09-09
|
||||
|
||||
内容
|
||||
----
|
||||
微信群列表对齐云客「微信群统计」页(2026-09-09 真机探明,数据源切
|
||||
``getChatroomList``):
|
||||
|
||||
* ``chat_group.group_remark`` —— 群备注列(getChatroomList.remark /
|
||||
friend_list/es 的 friendWxRemark,此前只存 raw_json);
|
||||
* ``chat_group_member_stats`` —— 群成员进出**按天**统计表。云客的新增/
|
||||
退出成员数是按任意时间窗实时算的,本地无成员进出事件流,改为每夜抓
|
||||
一次「当日窗」的 addMembers/lossMembers 落行,页面任意日窗=按天求和;
|
||||
上线日起积累,历史日期无数据。
|
||||
|
||||
幂等:列/表存在即跳过。SQLite/MySQL 通用。
|
||||
"""
|
||||
from collections.abc import Sequence
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = 'c9d0e1f2a3b4'
|
||||
down_revision: str | Sequence[str] | None = 'b7c8d9e0f1a2'
|
||||
branch_labels: str | Sequence[str] | None = None
|
||||
depends_on: str | Sequence[str] | None = None
|
||||
|
||||
_TABLE = 'chat_group_member_stats'
|
||||
_COL = 'group_remark'
|
||||
|
||||
|
||||
def _has_table(table: str) -> bool:
|
||||
insp = sa.inspect(op.get_bind())
|
||||
return table in insp.get_table_names()
|
||||
|
||||
|
||||
def _has_column(table: str, column: str) -> bool:
|
||||
insp = sa.inspect(op.get_bind())
|
||||
return column in {c["name"] for c in insp.get_columns(table)}
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
"""Upgrade schema."""
|
||||
if not _has_column("chat_group", _COL):
|
||||
op.add_column("chat_group", sa.Column(_COL, sa.String(255), nullable=True))
|
||||
# 回填:群备注从存量 raw_json.friendWxRemark 提取(friend_list/es 来源;
|
||||
# getChatroomList 行无备注键)。幂等:只补 NULL 行。MySQL 方言
|
||||
# (JSON_UNQUOTE),SQLite(测试库,不走生产迁移)跳过。
|
||||
if op.get_bind().dialect.name == "mysql":
|
||||
op.execute(
|
||||
"UPDATE chat_group SET group_remark = JSON_UNQUOTE(JSON_EXTRACT(raw_json, '$.friendWxRemark')) "
|
||||
"WHERE group_remark IS NULL "
|
||||
"AND JSON_UNQUOTE(JSON_EXTRACT(raw_json, '$.friendWxRemark')) IS NOT NULL "
|
||||
"AND JSON_UNQUOTE(JSON_EXTRACT(raw_json, '$.friendWxRemark')) != ''"
|
||||
)
|
||||
if _has_table(_TABLE):
|
||||
return # 已建过,幂等跳过
|
||||
op.create_table(
|
||||
_TABLE,
|
||||
sa.Column('channel', sa.String(16), primary_key=True),
|
||||
sa.Column('owner_account_id', sa.String(128), primary_key=True),
|
||||
sa.Column('room_id', sa.String(128), primary_key=True),
|
||||
sa.Column('day', sa.Integer(), primary_key=True),
|
||||
sa.Column('added', sa.Integer(), nullable=False, server_default='0'),
|
||||
sa.Column('lost', sa.Integer(), nullable=False, server_default='0'),
|
||||
sa.Column('fetched_at', sa.DateTime(), nullable=True),
|
||||
sa.Column('updated_at', sa.DateTime(), nullable=True),
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
"""Downgrade schema."""
|
||||
if _has_table(_TABLE):
|
||||
op.drop_table(_TABLE)
|
||||
if _has_column("chat_group", _COL):
|
||||
op.drop_column("chat_group", _COL)
|
||||
@@ -26,6 +26,12 @@ echo "$(date) --- 5/10 WeChat groups ---"
|
||||
$PY scripts/scrape_via_page.py --entity groups --max-employees 100
|
||||
echo "$(date) --- 5b/10 WA groups ---"
|
||||
$PY scripts/scrape_via_page.py --entity wa-groups
|
||||
# 5c: 微信群统计(getChatroomList,云客「微信群统计」同款)——全窗刷档案列
|
||||
# (真成员数/群备注/员工信息)+ 昨日窗落成员进出日行(页面窗内新增/退出)
|
||||
echo "$(date) --- 5c/10 WeChat chatroom stats ---"
|
||||
$PY scripts/scrape_via_page.py --entity chatroom-stats
|
||||
YDAY=$(date -d "yesterday" +%F)
|
||||
$PY scripts/scrape_via_page.py --entity chatroom-stats --day "$YDAY"
|
||||
echo "$(date) --- 6/10 Voice calls (WeChat + WA) ---"
|
||||
$PY scripts/scrape_via_page.py --entity voice-calls
|
||||
# 语音重扫会把 record_media_hash 覆盖回 None —— 必须紧跟录音重链接
|
||||
|
||||
@@ -1631,6 +1631,163 @@ async def scrape_wa_groups(cookie_str: str, max_pages: int = 200):
|
||||
return total_fetched, total_upserted
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 微信群统计 getChatroomList(2026-09-09 真机探明,云客同款数据源)
|
||||
#
|
||||
# 云客「沟通→微信号管理→微信群统计」页(/cms/wechat/wechatgroup)数据源:
|
||||
# POST /wechatweb/chatroom/getChatroomList(form-urlencoded)
|
||||
# departmentId + startTime/endTime('YYYY-MM-DD HH:MM:SS' 时间窗) + 分页;
|
||||
# 信封 data.overviews[] + totalCount(1,854 行,员工×群维度)。
|
||||
# 行含:真 memberCount(群成员数量——friend_list/es 的 friendNum 其实是
|
||||
# 员工好友数,旧映射存反了)、remark(群备注)、员工微信(wechatId=微信号
|
||||
# alias/wechatNickName/头像)、员工信息(userName/userPhone)、以及**窗内**
|
||||
# addMembers/lossMembers(新增/退出成员数,任意窗实时算)。
|
||||
#
|
||||
# 本地无成员进出事件流 → add/loss 只能按天积累:每夜抓「当日窗」落
|
||||
# chat_group_member_stats 日行,页面任意日窗=按天求和(历史无数据)。
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
CHATROOM_LIST_PATH = "/wechatweb/chatroom/getChatroomList"
|
||||
|
||||
|
||||
def _normalize_chatroom_stat_row(row: dict, owner_wxid: str) -> dict:
|
||||
"""getChatroomList 行 → ChatGroup 补列记录。
|
||||
|
||||
只放该端点有的字段——未出现的列不参与 upsert 的 UPDATE 子句,
|
||||
friend_list/es 抓的群主信息等旧值得以保留。
|
||||
"""
|
||||
rec = {
|
||||
"room_id": row.get("chatRoomId"),
|
||||
"channel": "wechat",
|
||||
"owner_account_id": owner_wxid,
|
||||
"room_name": row.get("chatRoomName"),
|
||||
# 真群成员数(修正:旧源 friendNum 是「员工好友数」被误存此列)
|
||||
"member_count": row.get("memberCount"),
|
||||
"head_image_url": row.get("headImageUrl"),
|
||||
"owner_account_nickname": row.get("wechatNickName"),
|
||||
"owner_user_id": row.get("userId"),
|
||||
"owner_user_phone": row.get("userPhone"),
|
||||
"message_count": row.get("messageCount"),
|
||||
"own_message_count": row.get("ownMessageCount"),
|
||||
"add_members": row.get("addMembers"), # 本窗快照(日窗=当日进出)
|
||||
"loss_members": row.get("lossMembers"),
|
||||
"last_chat_time": row.get("lastChatTime"),
|
||||
"update_time": row.get("updateTime"),
|
||||
"raw_json": row,
|
||||
}
|
||||
# 群备注:getChatroomList 行通常无 remark 键(云客页面也显示 -);有才写,
|
||||
# 无则不放该键——upsert 不更新未出现的列,friend_list/es 来源的
|
||||
# friendWxRemark 备注得以保留
|
||||
if row.get("remark"):
|
||||
rec["group_remark"] = row["remark"]
|
||||
return rec
|
||||
|
||||
|
||||
async def _chatroom_rows(page, dept: str, start_str: str, end_str: str, max_pages: int = 100):
|
||||
"""翻完 getChatroomList 一个窗,返回全部行(list)。"""
|
||||
rows_all: list[dict] = []
|
||||
page_num = 1
|
||||
while page_num <= max_pages:
|
||||
body = await _browser_fetch(
|
||||
page, CHATROOM_LIST_PATH,
|
||||
form_body={
|
||||
"departmentId": dept, "startTime": start_str, "endTime": end_str,
|
||||
"pageNum": page_num, "pageSize": 50,
|
||||
},
|
||||
)
|
||||
data = body.get("data") or {}
|
||||
rows = data.get("overviews") or []
|
||||
rows_all.extend(rows)
|
||||
total = data.get("totalCount") or 0
|
||||
logger.info(f" chatroom-list page {page_num}: {len(rows)} rows (total {total})")
|
||||
if not rows or page_num * 50 >= (total or 0):
|
||||
break
|
||||
page_num += 1
|
||||
await asyncio.sleep(1.2)
|
||||
return rows_all
|
||||
|
||||
|
||||
async def _owner_alias_map(conn):
|
||||
"""wechat_account 的 alias/nick_name → wx_id 映射。
|
||||
|
||||
getChatroomList 行的 wechatId 是员工**微信号**(alias,如 Nancyjade)
|
||||
而非 wxId,owner 归属需经此解析;miss 时回退存原值(不影响其余列)。
|
||||
"""
|
||||
from sqlalchemy import text as sa_text
|
||||
|
||||
result = await conn.execute(sa_text(
|
||||
"SELECT alias, nick_name, wx_id FROM wechat_account "
|
||||
"WHERE wx_id IS NOT NULL AND wx_id != ''"
|
||||
))
|
||||
mapping: dict[str, str] = {}
|
||||
for alias, nick, wx in result.fetchall():
|
||||
if alias:
|
||||
mapping[str(alias)] = wx
|
||||
if nick:
|
||||
mapping.setdefault(str(nick), wx)
|
||||
return mapping
|
||||
|
||||
|
||||
async def scrape_chatroom_stats(cookie_str: str, day: str | None = None):
|
||||
"""getChatroomList 抓取(云客微信群统计同款)。
|
||||
|
||||
* ``day=None``:全窗(2020 起)——chat_group 档案补列基线(真成员数/
|
||||
群备注/员工信息列;add/loss 存全窗快照);
|
||||
* ``day='YYYY-MM-DD'``:单日窗——落 chat_group_member_stats 当日行
|
||||
(页面任意日窗的新增/退出=按天求和)。
|
||||
"""
|
||||
import datetime as _dt
|
||||
|
||||
from app.db.base import async_engine
|
||||
from app.storage.repos import upsert_chat_group, upsert_chat_group_member_stats
|
||||
|
||||
dept = await _root_dept_id()
|
||||
async with async_engine.connect() as conn:
|
||||
alias_map = await _owner_alias_map(conn)
|
||||
|
||||
if day:
|
||||
start_str, end_str = f"{day} 00:00:00", f"{day} 23:59:59"
|
||||
day_key = int(day.replace("-", ""))
|
||||
else:
|
||||
start_str = "2020-01-01 00:00:00"
|
||||
end_str = f"{_dt.date.today()} 23:59:59"
|
||||
day_key = None
|
||||
|
||||
group_recs: list[dict] = []
|
||||
stats_recs: list[dict] = []
|
||||
async with PageScraper(cookie_str) as scraper:
|
||||
page = await _open_fetch_page(scraper)
|
||||
for row in await _chatroom_rows(page, dept, start_str, end_str):
|
||||
rid = row.get("chatRoomId")
|
||||
if not rid:
|
||||
continue
|
||||
owner_wxid = (
|
||||
alias_map.get(str(row.get("wechatId") or ""))
|
||||
or alias_map.get(str(row.get("wechatNickName") or ""))
|
||||
or str(row.get("wechatId") or "")
|
||||
)
|
||||
group_recs.append(_normalize_chatroom_stat_row(row, owner_wxid))
|
||||
if day_key is not None:
|
||||
stats_recs.append({
|
||||
"channel": "wechat",
|
||||
"owner_account_id": owner_wxid,
|
||||
"room_id": rid,
|
||||
"day": day_key,
|
||||
"added": row.get("addMembers") or 0,
|
||||
"lost": row.get("lossMembers") or 0,
|
||||
})
|
||||
|
||||
if group_recs:
|
||||
await _upsert_all(upsert_chat_group, group_recs)
|
||||
if stats_recs:
|
||||
await _upsert_all(upsert_chat_group_member_stats, stats_recs)
|
||||
logger.info(
|
||||
f"DONE: chatroom-stats day={day or '全窗'} groups={len(group_recs)} "
|
||||
f"stats_rows={len(stats_recs)}"
|
||||
)
|
||||
return len(group_recs), len(stats_recs)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 员工朋友圈 Moments (browser fetch, per employee)
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -2301,6 +2458,7 @@ async def main():
|
||||
"wa-accounts", # WA员工账号 (waBusiness/userlist interception)
|
||||
"wa-friends", # WA客户 (waBusiness/friendlist interception)
|
||||
"wa-groups", # WA群 (browser fetch businessFriendList type=1)
|
||||
"chatroom-stats", # 微信群统计 (getChatroomList 带窗,云客同款)
|
||||
"voice-calls", # 语音通话记录 (voice-call-list + wavoicecalllist)
|
||||
"voice-clips", # 长语音 (browser fetch call/list per employee)
|
||||
"wa-messages", # WA消息 (browser fetch chatRecords per conversation)
|
||||
@@ -2315,6 +2473,7 @@ async def main():
|
||||
parser.add_argument("--max-pages", type=int, default=200, help="列表翻页上限 (accounts/wa-accounts/wa-friends/voice-calls/moments)")
|
||||
parser.add_argument("--batch-size", type=int, default=50, help="媒体下载批次大小 (media)")
|
||||
parser.add_argument("--full-history", action="store_true", help="消息类实体强制全量翻历史(默认增量水位)")
|
||||
parser.add_argument("--day", type=str, default=None, help="chatroom-stats:单日窗 YYYY-MM-DD(落按天进出表);缺省=全窗档案基线")
|
||||
args = parser.parse_args()
|
||||
|
||||
cookie = _load_cookie()
|
||||
@@ -2348,6 +2507,8 @@ async def _dispatch_entity(args, cookie: str) -> None:
|
||||
await scrape_wa_friends(cookie, args.max_pages)
|
||||
elif args.entity == "wa-groups":
|
||||
await scrape_wa_groups(cookie, args.max_pages)
|
||||
elif args.entity == "chatroom-stats":
|
||||
await scrape_chatroom_stats(cookie, args.day)
|
||||
elif args.entity == "voice-calls":
|
||||
await scrape_voice_calls(cookie, args.max_pages)
|
||||
elif args.entity == "voice-clips":
|
||||
|
||||
@@ -344,3 +344,79 @@ async def test_q_room_name_substring_filter(make_client):
|
||||
# 空串 -> 不过滤
|
||||
r3 = await client.get("/api/groups", params={"q": ""})
|
||||
assert r3.json()["total"] == 3
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 微信群统计扩展(2026-09-09):统计时段窗聚合 + 员工列 + 群备注/最新消息筛选
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
async def _seed_stat(session: AsyncSession) -> None:
|
||||
from app.db.models import ChatGroupMemberStats, MsgDailyStats, Staff, WechatAccount
|
||||
|
||||
session.add_all([
|
||||
WechatAccount(wx_id="wx_owner_1", nick_name="玉", alias="Nancyjade",
|
||||
user_id="u_wx_1", raw_json={"headImageUrl": "http://a/h.png"}),
|
||||
Staff(user_id="u_wx_1", name="聂玉婷", phone="18500000000"),
|
||||
ChatGroup(room_id="stat_room@chatroom", channel="wechat", room_name="统计群",
|
||||
owner_account_id="wx_owner_1", owner_user_id="u_wx_1",
|
||||
member_count=46, group_remark="VIP群", last_chat_time=5000),
|
||||
# 窗内两天日行(0907+0908):沟通 3+4=7 条
|
||||
MsgDailyStats(channel="wechat", owner_account_id="wx_owner_1", peer_kind="room",
|
||||
peer_id="stat_room@chatroom", day=20260907, send_cnt=1, recv_cnt=2),
|
||||
MsgDailyStats(channel="wechat", owner_account_id="wx_owner_1", peer_kind="room",
|
||||
peer_id="stat_room@chatroom", day=20260908, send_cnt=2, recv_cnt=2),
|
||||
# 窗外日行(0901)不应计入
|
||||
MsgDailyStats(channel="wechat", owner_account_id="wx_owner_1", peer_kind="room",
|
||||
peer_id="stat_room@chatroom", day=20260901, send_cnt=99, recv_cnt=99),
|
||||
# 窗内进出:0907 +2/-1
|
||||
ChatGroupMemberStats(channel="wechat", owner_account_id="wx_owner_1",
|
||||
room_id="stat_room@chatroom", day=20260907, added=2, lost=1),
|
||||
])
|
||||
await session.commit()
|
||||
|
||||
|
||||
async def test_group_window_aggregates_and_staff_columns(make_client):
|
||||
"""msg_start/msg_end 窗内三聚合 + 员工微信/员工信息关联列。"""
|
||||
client = await make_client(_seed_stat)
|
||||
# 2026-09-07 ~ 2026-09-09(上海日 20260907..20260909,UTC ms)
|
||||
r = await client.get("/api/groups", params={
|
||||
"channel": "wechat", "room_id": "stat_room@chatroom",
|
||||
"msg_start": 1788758400000, "msg_end": 1789017600000,
|
||||
})
|
||||
assert r.status_code == 200
|
||||
it = r.json()["items"][0]
|
||||
assert it["window_msg_count"] == 7 # 窗内日行求和,0901 排除
|
||||
assert it["window_added"] == 2
|
||||
assert it["window_lost"] == 1
|
||||
assert it["group_remark"] == "VIP群"
|
||||
assert it["staff_nickname"] == "玉"
|
||||
assert it["staff_wx_alias"] == "Nancyjade"
|
||||
assert it["staff_avatar"] == "http://a/h.png"
|
||||
assert it["staff_name"] == "聂玉婷"
|
||||
assert it["staff_phone"] == "18500000000"
|
||||
|
||||
# 无窗请求 → 聚合三列为 None(不误显 0)
|
||||
r2 = await client.get("/api/groups", params={
|
||||
"channel": "wechat", "room_id": "stat_room@chatroom"})
|
||||
it2 = r2.json()["items"][0]
|
||||
assert it2["window_msg_count"] is None
|
||||
assert it2["window_added"] is None
|
||||
|
||||
|
||||
async def test_group_remark_and_last_window_filters(make_client):
|
||||
"""群备注子串 + 最新消息时间段过滤。"""
|
||||
client = await make_client(_seed_stat)
|
||||
r = await client.get("/api/groups", params={
|
||||
"channel": "wechat", "group_remark": "VIP"})
|
||||
assert [i["room_id"] for i in r.json()["items"]] == ["stat_room@chatroom"]
|
||||
r2 = await client.get("/api/groups", params={
|
||||
"channel": "wechat", "group_remark": "不存在"})
|
||||
assert r2.json()["total"] == 0
|
||||
|
||||
# last_chat_time=5000 → last_start=4000 命中,last_start=6000 排除
|
||||
r3 = await client.get("/api/groups", params={
|
||||
"channel": "wechat", "last_start": 4000, "last_end": 9999})
|
||||
assert "stat_room@chatroom" in [i["room_id"] for i in r3.json()["items"]]
|
||||
r4 = await client.get("/api/groups", params={
|
||||
"channel": "wechat", "last_start": 6000, "last_end": 9999})
|
||||
assert r4.json()["total"] == 0
|
||||
|
||||
Reference in New Issue
Block a user