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

feat(db): moments_post 表+upsert(朋友圈实体,PK=云客data.id,create_time统一ms)

Browse Source 
- MomentsPost 模型:moments_post 表,PK=id(云客 getFriendsterInfo data.id),
  wechat_id/create_time 建索引,praise/comment/image 存 JSON 列
- 迁移 524da8d2d597(手写 create_table,已应用到 smoke.db 并在临时 SQLite
  全链路 upgrade/downgrade 验证)
- repos.upsert_moments_posts:批量幂等 upsert;附带修 _build_column_values
  对 JSON 列表的误 join(标量列 join 保持不变,dept_route 旧测试仍过)
- 修复 3 个 chat_group 旧测试缺 owner_account_id(2026-08-24 复合主键遗留)
- 全套 430 测试通过

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
浅念
2026-08-25 13:44:54 +08:00
parent 2f73d79a6d
commit b7669199f8
5 changed files with 221 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yunke-server/app/db/models.py
+29 -1
View File
@@ -26,7 +26,7 @@ Conventions
- ``SyncState`` uses ``channel`` to scope a sync cursor (together with
``entity_type``).
- ``Staff``, ``Department``, ``WechatAccount``, ``WhatsappAccount``,
``YkAccount``, ``SyncRun`` carry no ``channel`` column.
``YkAccount``, ``SyncRun``, ``MomentsPost`` carry no ``channel`` column.
* Primary keys use the natural source id where one exists; otherwise an
integer surrogate PK is used (``SyncState``, ``YkAccount``, ``SyncRun``).
* ``GroupMember`` uses a composite PK ``(channel, room_id, member_id)``.
@@ -592,6 +592,33 @@ class Sms(Base):
__table_args__ = (Index("ix_sms_user_time", "user_id", "create_time"),)
class MomentsPost(Base):
"""员工朋友圈帖子 — source: ``POST /wechatweb/wechatFriendster/getFriendsterInfo``.
PK = 云客 ``data.id``。``create_time`` 统一 **ms epoch**（云客返回秒，
scraper ×1000）。WeChat-only 实体（无 ``channel`` 列），所属员工微信号
由 ``wechat_id`` 标识。
"""
__tablename__ = "moments_post"
id: Mapped[str] = mapped_column(String(64), primary_key=True)
wechat_id: Mapped[str] = mapped_column(String(128), index=True)
sns_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
author_name: Mapped[str | None] = mapped_column(String(255), nullable=True)
author_wxid: Mapped[str | None] = mapped_column(String(128), nullable=True)
author_avatar: Mapped[str | None] = mapped_column(String(1024), nullable=True)
content: Mapped[str | None] = mapped_column(Text, nullable=True)
type_raw: Mapped[str | None] = mapped_column(String(16), nullable=True)
create_time: Mapped[int | None] = mapped_column(BigInteger, index=True)
praise_json: Mapped[list[dict[str, Any]] | None] = mapped_column(JSON, nullable=True)
comment_json: Mapped[list[dict[str, Any]] | None] = mapped_column(JSON, nullable=True)
image_urls: Mapped[list[str] | None] = mapped_column(JSON, nullable=True)
raw_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
fetched_at: Mapped[dt.datetime] = _fetched_at()
updated_at: Mapped[dt.datetime] = _updated_at()
# ===========================================================================
# 5. Operational / bookkeeping tables (local; no raw_json/version)
# ===========================================================================
@@ -696,6 +723,7 @@ __all__ = [
"GroupMember",
"Media",
"Message",
"MomentsPost",
"PhoneCall",
"Sms",
"Staff",
yunke-server/app/storage/repos.py
+38 -5
View File
@@ -36,6 +36,10 @@ dict (see :func:`_coerce_bool`, :func:`_coerce_int`):
* ``raw_json``: the original record dict, preserved verbatim. When the input
already carries a ``raw_json`` key (scraper output does), that value wins;
otherwise the input dict itself is stored.
* JSON-typed list columns (``MomentsPost.praise_json`` / ``comment_json`` /
``image_urls``): list values are kept verbatim (stored as JSON arrays). The
list -> ``";"``-join coercion only applies to lists headed for SCALAR
String columns (e.g. ``Customer.dept_route``).
Sharding note
-------------
@@ -60,6 +64,7 @@ from app.db.models import (
GroupMember,
Media,
Message,
MomentsPost,
PhoneCall,
Sms,
Staff,
@@ -75,6 +80,7 @@ __all__ = [
"upsert_department",
"upsert_media",
"upsert_messages",
"upsert_moments_posts",
"upsert_phone_call",
"upsert_sms",
"upsert_staff",
@@ -220,12 +226,16 @@ def _build_column_values(
on unknown columns).
* Boolean / int-as-str fields are coerced via :func:`_coerce_bool` /
:func:`_coerce_int`.
* List values headed for SCALAR columns (e.g. ``Customer.dept_route``) are
joined into a ``";"``-separated string so the bind succeeds; list values
headed for JSON columns (e.g. ``MomentsPost.praise_json`` /
``image_urls``) are kept verbatim as JSON arrays.
* ``raw_json`` defaults to the entire input dict when not explicitly
provided, so the verbatim upstream record is always recoverable.
"""
col_names = {c.name for c in model.__table__.columns}
cols = {c.name: c for c in model.__table__.columns}
values: dict[str, Any] = {}
for col_name in col_names:
for col_name, col in cols.items():
if col_name in rec:
v = rec[col_name]
if col_name in _BOOL_FIELDS:
@@ -235,14 +245,15 @@ def _build_column_values(
# Some upstream fields are lists (e.g. Customer.deptRoute is a
# list of dept ids) but map to scalar String columns — join so the
# bind succeeds and the data is still recoverable (raw_json keeps
# the original list).
if isinstance(v, list):
# the original list). JSON-typed columns keep the list verbatim
# (MomentsPost.praise_json / comment_json / image_urls).
if isinstance(v, list) and not isinstance(col.type, sa.JSON):
v = ";".join(str(x) for x in v)
values[col_name] = v
# Preserve the verbatim source record. If the scraper already set raw_json
# (it does), that wins; otherwise the input dict itself is the source of
# truth.
if "raw_json" in col_names and "raw_json" not in values:
if "raw_json" in cols and "raw_json" not in values:
values["raw_json"] = rec
# Sanitize lone surrogate chars (e.g. \udf1e from emoji fragments in the
# upstream API) that break SQLite/MySQL UTF-8 encoding. Replaces them with
@@ -602,6 +613,28 @@ async def upsert_sms(session: AsyncSession, rec: dict[str, Any]) -> None:
await _upsert_one(session, Sms, rec)
# ---------------------------------------------------------------------------
# MomentsPost (员工朋友圈) upsert — 员工聊天窗口 Phase 1
# ---------------------------------------------------------------------------
async def upsert_moments_posts(session: AsyncSession, recs: list[dict[str, Any]]) -> int:
"""Batch idempotent upsert for :class:`MomentsPost` (PK = ``id``).
Same ``id`` twice -> one row, with non-PK fields refreshed from the later
record. ``create_time`` goes through the ``_INT_FIELDS`` int coercion (it
must land as a ms-epoch int; 云客 returns seconds, the scraper ×1000).
JSON list columns (``praise_json`` / ``comment_json`` / ``image_urls``)
are stored verbatim as JSON arrays — the generic list->";".join coercion
in :func:`_build_column_values` only applies to scalar columns. Returns
the number of records processed (input length); the DB collapses
duplicate ``id`` values to one row.
"""
for rec in recs:
await _upsert_one(session, MomentsPost, rec)
return len(recs)
# ---------------------------------------------------------------------------
# VoiceClip (长语音) upsert
# ---------------------------------------------------------------------------
yunke-server/migrations/versions/524da8d2d597_moments_post_table.py
+49
View File
@@ -0,0 +1,49 @@
"""moments_post table
Revision ID: 524da8d2d597
Revises: d8eee96cd017
Create Date: 2026-08-25 13:38:31.301937
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = '524da8d2d597'
down_revision: str | Sequence[str] | None = 'd8eee96cd017'
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
"""Upgrade schema."""
op.create_table('moments_post',
sa.Column('id', sa.String(length=64), nullable=False),
sa.Column('wechat_id', sa.String(length=128), nullable=False),
sa.Column('sns_id', sa.String(length=64), nullable=True),
sa.Column('author_name', sa.String(length=255), nullable=True),
sa.Column('author_wxid', sa.String(length=128), nullable=True),
sa.Column('author_avatar', sa.String(length=1024), nullable=True),
sa.Column('content', sa.Text(), nullable=True),
sa.Column('type_raw', sa.String(length=16), nullable=True),
sa.Column('create_time', sa.BigInteger(), nullable=True),
sa.Column('praise_json', sa.JSON(), nullable=True),
sa.Column('comment_json', sa.JSON(), nullable=True),
sa.Column('image_urls', sa.JSON(), nullable=True),
sa.Column('raw_json', sa.JSON(), nullable=True),
sa.Column('fetched_at', sa.DateTime(), server_default=sa.func.now(), nullable=False),
sa.Column('updated_at', sa.DateTime(), server_default=sa.func.now(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_index('ix_moments_post_wechat_id', 'moments_post', ['wechat_id'], unique=False)
op.create_index('ix_moments_post_create_time', 'moments_post', ['create_time'], unique=False)
def downgrade() -> None:
"""Downgrade schema."""
op.drop_index('ix_moments_post_create_time', table_name='moments_post')
op.drop_index('ix_moments_post_wechat_id', table_name='moments_post')
op.drop_table('moments_post')
yunke-server/tests/test_models.py
+48
View File
@@ -269,3 +269,51 @@ async def test_group_member_composite_pk():
assert fetched.nickname == "Bob"
finally:
await engine.dispose()
async def test_moments_post_round_trip():
"""A MomentsPost row round-trips JSON list columns + ms create_time."""
from app.db.models import MomentsPost
engine = await _make_engine()
try:
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
async with SessionLocal() as session:
row = MomentsPost(
id="69e082a3c3524a7fe3ac53b4",
wechat_id="wxid_x",
sns_id="-3546042415332912522",
author_name="Yoyo",
author_wxid="wxid_x",
author_avatar="https://wx.qlogo.cn/avatars/yoyo",
content="Welcome to HESU",
type_raw="1",
create_time=1776302051000,
praise_json=[{"name": "E**U"}],
comment_json=[],
image_urls=["https://wx.oss.example/a.jpg"],
raw_json={"id": "69e082a3c3524a7fe3ac53b4"},
)
session.add(row)
await session.commit()
fetched = (
await session.execute(
sa.select(MomentsPost).where(
MomentsPost.id == "69e082a3c3524a7fe3ac53b4"
)
)
).scalar_one()
assert fetched.wechat_id == "wxid_x"
assert fetched.sns_id == "-3546042415332912522"
# JSON list columns must round-trip as lists (not joined strings).
assert fetched.praise_json[0]["name"] == "E**U"
assert fetched.comment_json == []
assert fetched.image_urls == ["https://wx.oss.example/a.jpg"]
# create_time stays the ms-epoch int exactly as written.
assert fetched.create_time == 1776302051000
assert fetched.type_raw == "1"
assert fetched.raw_json == {"id": "69e082a3c3524a7fe3ac53b4"}
assert fetched.fetched_at is not None
finally:
await engine.dispose()
yunke-server/tests/test_repos.py
+57
View File
@@ -28,6 +28,7 @@ from app.db.models import (
Department,
GroupMember,
Message,
MomentsPost,
PhoneCall,
Sms,
Staff,
@@ -42,6 +43,7 @@ from app.storage.repos import (
upsert_customers,
upsert_department,
upsert_messages,
upsert_moments_posts,
upsert_phone_call,
upsert_sms,
upsert_staff,
@@ -820,6 +822,7 @@ async def test_upsert_chat_group_reupsert_updates_group_keeps_members_unique():
rec1 = {
"room_id": "98765432109@chatroom",
"channel": "wechat",
"owner_account_id": "wxid_owner_re",
"room_name": "Old Name",
"member_count": 2,
"member_ids": "wxid_m1;wxid_m2",
@@ -877,6 +880,7 @@ async def test_upsert_chat_group_drops_empty_member_id():
rec = {
"room_id": "55555555555@chatroom",
"channel": "wechat",
"owner_account_id": "wxid_owner_empty",
# Empty string sneaks through; repo must skip it.
"member_ids_list": ["", "wxid_only", ""],
}
@@ -905,6 +909,7 @@ async def test_upsert_chat_group_with_no_members_writes_group_only():
rec = {
"room_id": "empty@chatroom",
"channel": "wechat",
"owner_account_id": "wxid_owner_none",
"room_name": "No Members Yet",
"member_ids_list": [],
}
@@ -1402,3 +1407,55 @@ async def test_upsert_sms_coerces_direction_and_create_time():
assert fetched.content == "hi"
finally:
await engine.dispose()
# ---------------------------------------------------------------------------
# MomentsPost upsert (员工聊天窗口 Phase 1)
# ---------------------------------------------------------------------------
async def test_upsert_moments_posts_idempotent_keeps_json_lists_and_coerces_create_time():
"""MomentsPost batch upsert: idempotent on PK ``id``; JSON list columns
(praise_json / comment_json / image_urls) stay real lists (NOT joined into
";..." strings by the scalar-column list coercion); numeric-str
create_time coerces to ms-epoch int.
"""
engine = await _make_engine()
try:
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
async with SessionLocal() as session:
rec1 = {
"id": "69e082a3c3524a7fe3ac53b4",
"wechat_id": "wxid_x",
"sns_id": "-3546042415332912522",
"author_name": "Yoyo",
"author_wxid": "wxid_x",
"content": "v1 content",
"type_raw": "1",
"create_time": 1776302051000, # ms epoch int
"praise_json": [{"name": "E**U"}],
"comment_json": [],
"image_urls": ["https://wx.oss.example/a.jpg"],
}
# Same PK again: later write wins, create_time arrives as a
# numeric string (defensive coercion path).
rec2 = dict(rec1, content="v2 content", create_time="1776302051999")
count = await upsert_moments_posts(session, [rec1, rec2])
await session.commit()
assert count == 2 # returns input length
rows = (
await session.execute(
sa.select(MomentsPost).execution_options(populate_existing=True)
)
).scalars().all()
assert len(rows) == 1 # same PK twice -> ONE row
assert rows[0].content == "v2 content" # later record wins
assert rows[0].create_time == 1776302051999 # str -> int
# JSON list columns preserved verbatim as lists.
assert rows[0].praise_json == [{"name": "E**U"}]
assert rows[0].comment_json == []
assert rows[0].image_urls == ["https://wx.oss.example/a.jpg"]
assert rows[0].type_raw == "1" # MomentsPost type_raw is a string code
finally:
await engine.dispose()