diff --git a/yunke-server/app/db/models.py b/yunke-server/app/db/models.py index 8d5a249..bca1141 100644 --- a/yunke-server/app/db/models.py +++ b/yunke-server/app/db/models.py @@ -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", diff --git a/yunke-server/app/storage/repos.py b/yunke-server/app/storage/repos.py index 41eaaa5..e3ddf28 100644 --- a/yunke-server/app/storage/repos.py +++ b/yunke-server/app/storage/repos.py @@ -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 # --------------------------------------------------------------------------- diff --git a/yunke-server/migrations/versions/524da8d2d597_moments_post_table.py b/yunke-server/migrations/versions/524da8d2d597_moments_post_table.py new file mode 100644 index 0000000..bbba26c --- /dev/null +++ b/yunke-server/migrations/versions/524da8d2d597_moments_post_table.py @@ -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') diff --git a/yunke-server/tests/test_models.py b/yunke-server/tests/test_models.py index d1be010..67d3b8a 100644 --- a/yunke-server/tests/test_models.py +++ b/yunke-server/tests/test_models.py @@ -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() diff --git a/yunke-server/tests/test_repos.py b/yunke-server/tests/test_repos.py index 529d288..95c9f36 100644 --- a/yunke-server/tests/test_repos.py +++ b/yunke-server/tests/test_repos.py @@ -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()