diff --git a/yunke-server/app/db/models.py b/yunke-server/app/db/models.py index 9906312..69cd801 100644 --- a/yunke-server/app/db/models.py +++ b/yunke-server/app/db/models.py @@ -351,7 +351,12 @@ class Message(Base): String(128), nullable=True ) # @chatroom id; NULL/empty = direct message is_send: Mapped[bool | None] = mapped_column(Boolean, nullable=True) # mine/isSend - type: Mapped[int | None] = mapped_column(Integer, nullable=True) + type: Mapped[str | None] = mapped_column( + String(32), nullable=True + ) # normalized enum label (text/image/voice/.../unknown); see design §4.3 + type_raw: Mapped[int | None] = mapped_column( + Integer, nullable=True + ) # original upstream type code (e.g. WeChat 1=text, -2=divider) text: Mapped[str | None] = mapped_column(Text, nullable=True) # content/text file_url: Mapped[str | None] = mapped_column(String(1024), nullable=True) file_th_url: Mapped[str | None] = mapped_column(String(1024), nullable=True) diff --git a/yunke-server/app/scraper/wechat.py b/yunke-server/app/scraper/wechat.py index bcf6e65..25ad6ca 100644 --- a/yunke-server/app/scraper/wechat.py +++ b/yunke-server/app/scraper/wechat.py @@ -190,6 +190,34 @@ async def scrape_wechat_friends( # =========================================================================== +def _coerce_int_or_none(value: Any) -> int | None: + """Coerce a possibly-string numeric value to ``int``; ``None`` preserved. + + Mirrors :func:`app.storage.repos._coerce_int` so the scraper can emit a + ready-to-store ``type_raw`` int and check raw type codes (e.g. the ``-2`` + divider drop) without depending on the storage layer (the scraper has no + DB dependency by design). A non-numeric value (``"text"`` / ``None`` / + missing) returns ``None``. + """ + if value is None: + return None + if isinstance(value, bool): + return int(value) + if isinstance(value, int): + return value + if isinstance(value, float): + return int(value) + if isinstance(value, str): + s = value.strip() + if not s: + return None + try: + return int(s) + except ValueError: + return None + return None + + async def _fetch_msg_page( client: YunkeClient, *, @@ -245,9 +273,13 @@ def _normalize_message(record: dict[str, Any]) -> dict[str, Any]: """Map one upstream ``msg/chats`` record to the ``Message`` column shape. Adds ``channel="wechat"`` and preserves the original record under - ``raw_json``. The ``type`` field is normalized via - :func:`normalize_wechat_type` to a stable enum string; the raw code is - still recoverable from ``raw_json`` for the persistence layer. + ``raw_json``. Per design §4.3 the message type is carried as TWO fields: + + * ``type`` — the normalized enum label (``"text"`` / ``"unknown"`` / ...) + via :func:`normalize_wechat_type`. + * ``type_raw`` — the original upstream type code (``messageType`` falling + back to ``type``), coerced to ``int | None`` so the raw code survives + end-to-end without round-tripping through ``raw_json``. ``None``-safe: missing fields become ``None`` rather than raising, so a partial payload does not break the whole scrape. Value coercion @@ -281,6 +313,7 @@ def _normalize_message(record: dict[str, Any]) -> dict[str, Any]: "room_id": record.get("roomid"), "is_send": is_send, "type": normalize_wechat_type(raw_type), + "type_raw": _coerce_int_or_none(raw_type), "text": record.get("text"), "file_url": record.get("file"), "file_th_url": record.get("fileTh"), @@ -305,15 +338,19 @@ def _normalize_message(record: dict[str, Any]) -> dict[str, Any]: def _is_divider(record: dict[str, Any]) -> bool: - """True iff this record is a date-divider (``type``/``messageType`` == -2). + """True iff this record is a date-divider (raw type code ``-2``). These are not real messages — the upstream inserts them as visual date bars in the chat UI — and must never be yielded or persisted. Per the - Task 1.6 spec we drop a record when EITHER ``type`` or ``messageType`` is + Task 1.6 spec we drop a record when EITHER ``messageType`` or ``type`` is ``-2``; all other unknown codes pass through. + + The check is on the RAW code (not the normalized label) so the drop + decision is robust against any future change to + :func:`normalize_wechat_type`'s label vocabulary. """ - return normalize_wechat_type(record.get("messageType")) == "divider" or ( - normalize_wechat_type(record.get("type")) == "divider" + return _coerce_int_or_none(record.get("messageType")) == -2 or ( + _coerce_int_or_none(record.get("type")) == -2 ) diff --git a/yunke-server/app/storage/repos.py b/yunke-server/app/storage/repos.py index f430b4d..3981985 100644 --- a/yunke-server/app/storage/repos.py +++ b/yunke-server/app/storage/repos.py @@ -23,9 +23,9 @@ dict (see :func:`_coerce_bool`, :func:`_coerce_int`): ``1`` / ``"1"`` / ``True`` -> ``True``; ``0`` / ``"0"`` / ``False`` -> ``False``; ``None`` preserved. * ``gender``: ``"0"`` / ``"1"`` / ``"2"`` (str) -> ``int`` when not ``None``. -* ``type`` (Message): coerced to ``int``; a non-numeric string (e.g. a - normalized type label like ``"text"``) collapses to ``None`` — the raw code - is still recoverable from ``raw_json``. +* ``type_raw`` (Message): the raw upstream type code, coerced to + ``int | None``. ``Message.type`` is a normalized label string and is + stored as-is (no int coercion). * Timestamps (``last_chat_time`` / ``timestamp`` / ``last_update`` / ...): passed through as-is (ms-epoch ``int`` or ``None``). * ``raw_json``: the original record dict, preserved verbatim. When the input @@ -72,7 +72,7 @@ _BOOL_FIELDS: frozenset[str] = frozenset( _INT_FIELDS: frozenset[str] = frozenset( { "gender", - "type", + "type_raw", "pic_type", "friend_state", "delete_flag", @@ -118,8 +118,7 @@ def _coerce_int(value: Any) -> int | None: ``"2"`` -> ``2``; ``2`` -> ``2``; ``0``/``"0"`` -> ``0``; ``None`` -> ``None``. A non-numeric string (e.g. ``"text"``) returns ``None`` — the - caller (Message.type) treats that as "no value", and the raw code is still - recoverable from ``raw_json``. + caller (e.g. Message.type_raw) treats that as "no value". """ if value is None: return None @@ -287,9 +286,10 @@ async def upsert_customers(session: AsyncSession, recs: list[dict[str, Any]]) -> async def upsert_messages(session: AsyncSession, recs: list[dict[str, Any]]) -> int: """Batch idempotent upsert for :class:`Message` (PK = ``msg_id``). - Applies the documented coercions (``is_send``/``is_del`` -> bool, ``type`` - -> int). Returns the number of records processed (input length); the DB - collapses duplicate ``msg_id`` values to one row. + Applies the documented coercions (``is_send``/``is_del`` -> bool, + ``type_raw`` -> int). ``Message.type`` is the normalized label string and + is stored as-is. Returns the number of records processed (input length); + the DB collapses duplicate ``msg_id`` values to one row. .. note:: For M1 the ``Message`` model maps to a single physical ``message`` diff --git a/yunke-server/migrations/env.py b/yunke-server/migrations/env.py index 64cba06..785f00e 100644 --- a/yunke-server/migrations/env.py +++ b/yunke-server/migrations/env.py @@ -50,6 +50,11 @@ def _sync_url(url: str) -> str: def run_migrations_offline() -> None: """Run migrations in 'offline' mode (emit SQL without a live connection).""" url = _sync_url(get_settings().db_url) + # SQLite cannot ALTER most column properties directly; Alembic's batch mode + # (table-recreate) handles it. Enable it for SQLite so the SAME migration + # scripts run on both SQLite (dev/test) and MySQL (prod). MySQL needs no + # batch mode and gets direct ALTERs. + render_as_batch = url.startswith("sqlite") context.configure( url=url, target_metadata=target_metadata, @@ -57,6 +62,7 @@ def run_migrations_offline() -> None: dialect_opts={"paramstyle": "named"}, compare_type=True, compare_server_default=True, + render_as_batch=render_as_batch, ) with context.begin_transaction(): @@ -74,11 +80,14 @@ def run_migrations_online() -> None: ) with connectable.connect() as connection: + # See offline mode for the rationale on batch mode per dialect. + render_as_batch = connection.dialect.name == "sqlite" context.configure( connection=connection, target_metadata=target_metadata, compare_type=True, compare_server_default=True, + render_as_batch=render_as_batch, ) with context.begin_transaction(): diff --git a/yunke-server/migrations/versions/96975050fa79_message_type_split_normalized_label_and_.py b/yunke-server/migrations/versions/96975050fa79_message_type_split_normalized_label_and_.py new file mode 100644 index 0000000..073d977 --- /dev/null +++ b/yunke-server/migrations/versions/96975050fa79_message_type_split_normalized_label_and_.py @@ -0,0 +1,59 @@ +"""message type: split normalized label and raw code + +Revision ID: 96975050fa79 +Revises: d7a9bc760c6b +Create Date: 2026-08-01 12:01:31.810063 + +Splits the ``message.type`` column to match design §4.3: + +* ``type`` becomes ``String(32)`` holding the NORMALIZED enum label produced + by :func:`app.scraper.parsers.normalize_wechat_type` (``"text"`` / + ``"unknown"`` / ...). Previously this column was ``Integer`` but the + scraper was already writing a normalized string into it, so the storage + layer silently coerced that string to ``NULL`` and the only way to recover + the raw code was via ``raw_json``. +* ``type_raw`` (new, ``Integer``) carries the ORIGINAL upstream type code + end-to-end (scraper -> repos -> DB) so it no longer has to be round-tripped + through ``raw_json``. + +The operations are wrapped in ``batch_alter_table`` so the SAME migration +runs on SQLite (dev/test, where ALTER COLUMN is unsupported and Alembic +recreates the table) and MySQL (prod, where it emits a direct ``MODIFY +COLUMN`` / ``ADD COLUMN``). +""" +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = '96975050fa79' +down_revision: str | Sequence[str] | None = 'd7a9bc760c6b' +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + with op.batch_alter_table('message', schema=None) as batch_op: + batch_op.add_column(sa.Column('type_raw', sa.Integer(), nullable=True)) + batch_op.alter_column('type', + existing_type=sa.INTEGER(), + type_=sa.String(length=32), + existing_nullable=True) + + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + with op.batch_alter_table('message', schema=None) as batch_op: + batch_op.alter_column('type', + existing_type=sa.String(length=32), + type_=sa.INTEGER(), + existing_nullable=True) + batch_op.drop_column('type_raw') + + # ### end Alembic commands ### diff --git a/yunke-server/tests/test_models.py b/yunke-server/tests/test_models.py index 2b4f8a9..d1be010 100644 --- a/yunke-server/tests/test_models.py +++ b/yunke-server/tests/test_models.py @@ -179,7 +179,8 @@ async def test_message_natural_pk(): owner_account_id="wxid_staff", talker_id="wxid_abc", is_send=False, - type=1, + type="text", + type_raw=1, text="hello", timestamp=1700000000000, ) @@ -195,6 +196,8 @@ async def test_message_natural_pk(): assert fetched.text == "hello" assert fetched.timestamp == 1700000000000 assert fetched.is_send is False + assert fetched.type == "text" + assert fetched.type_raw == 1 finally: await engine.dispose() diff --git a/yunke-server/tests/test_repos.py b/yunke-server/tests/test_repos.py index cac2721..8ff77cb 100644 --- a/yunke-server/tests/test_repos.py +++ b/yunke-server/tests/test_repos.py @@ -72,10 +72,10 @@ def test_coerce_int_handles_documented_cases(): def test_coerce_int_returns_none_for_non_numeric_string(): - """A non-numeric string (e.g. normalized 'type' label) -> None. + """A non-numeric string (e.g. a label) -> None. - The column is INTEGER; a non-numeric value cannot be stored. The raw value - remains recoverable from ``raw_json``. + Integer columns (``type_raw``, ``gender``, ...) cannot store a + non-numeric value; the coercion collapses it to ``None``. """ assert _coerce_int("text") is None assert _coerce_int("unknown") is None @@ -229,7 +229,8 @@ async def test_upsert_customers_batch_returns_count(): async def test_upsert_messages_coerces_types(): - """is_send '1' -> True; is_del 0 -> False; same msg_id twice -> 1 row.""" + """is_send '1' -> True; is_del 0 -> False; type label stored as string; + type_raw coerced to int; same msg_id twice -> 1 row.""" engine = await _make_engine() try: SessionLocal = async_sessionmaker(engine, expire_on_commit=False) @@ -239,7 +240,8 @@ async def test_upsert_messages_coerces_types(): "channel": "wechat", "is_send": "1", "is_del": 0, - "type": "1", + "type": "text", # normalized label — stored as-is (no coercion) + "type_raw": "1", # raw code arrives as str -> coerced to int "timestamp": 1_700_000_000_000, } await upsert_messages(session, [rec]) @@ -252,17 +254,18 @@ async def test_upsert_messages_coerces_types(): ).scalar_one() assert fetched.is_send is True assert fetched.is_del is False - assert fetched.type == 1 + assert fetched.type == "text" # string label preserved + assert fetched.type_raw == 1 # raw code coerced to int assert fetched.timestamp == 1_700_000_000_000 - # Upsert same msg_id again with different text -> row updated, not - # duplicated. - rec2 = dict(rec, text="updated") + # Upsert same msg_id again with different text + new raw code -> + # row updated, not duplicated. + rec2 = dict(rec, text="updated", type="unknown", type_raw=3) await upsert_messages(session, [rec2]) await session.commit() # ``populate_existing=True`` bypasses the identity-map cache so we - # observe the second upsert's text value. + # observe the second upsert's values. rows = ( await session.execute( sa.select(Message) @@ -272,6 +275,34 @@ async def test_upsert_messages_coerces_types(): ).scalars().all() assert len(rows) == 1 assert rows[0].text == "updated" + assert rows[0].type == "unknown" + assert rows[0].type_raw == 3 + finally: + await engine.dispose() + + +async def test_upsert_messages_preserves_none_type_and_type_raw(): + """Missing / None type & type_raw stay None (no spurious coercion).""" + engine = await _make_engine() + try: + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as session: + rec = { + "msg_id": "m_none", + "channel": "wechat", + "type": None, + "type_raw": None, + } + await upsert_messages(session, [rec]) + await session.commit() + + fetched = ( + await session.execute( + sa.select(Message).where(Message.msg_id == "m_none") + ) + ).scalar_one() + assert fetched.type is None + assert fetched.type_raw is None finally: await engine.dispose() diff --git a/yunke-server/tests/test_scrape_wechat_messages.py b/yunke-server/tests/test_scrape_wechat_messages.py index 48cbaf8..cd17dfa 100644 --- a/yunke-server/tests/test_scrape_wechat_messages.py +++ b/yunke-server/tests/test_scrape_wechat_messages.py @@ -106,7 +106,8 @@ async def test_scrape_wechat_messages_drops_divider_and_paginates_backward(): assert r1["ori_talker"] == "wxid_friend_0001" assert r1["room_id"] == "" # raw value carried through assert r1["is_send"] == 0 # pulled from `mine` - assert r1["type"] == "text" + assert r1["type"] == "text" # normalized label (design §4.3) + assert r1["type_raw"] == 1 # original raw code preserved assert r1["text"] == "hi alice" assert r1["timestamp"] == E1 + 600000 assert r1["sender_name"] == "Alice Wang" @@ -119,6 +120,7 @@ async def test_scrape_wechat_messages_drops_divider_and_paginates_backward(): assert r2["ori_talker"] == "room12345@chatroom" assert r2["is_send"] == 1 assert r2["type"] == "unknown" # unmapped code passes through + assert r2["type_raw"] == 3 # raw image-type code preserved assert r2["file_url"] == "https://cdn.example.com/img2.jpg" assert r2["file_th_url"] == "https://cdn.example.com/img2_th.jpg" assert r2["file_hd_url"] == "https://cdn.example.com/img2_hd.jpg" @@ -133,6 +135,7 @@ async def test_scrape_wechat_messages_drops_divider_and_paginates_backward(): # m3 — voice / transcription. assert r3["msg_id"] == "999999" assert r3["type"] == "unknown" + assert r3["type_raw"] == 34 # raw voice-type code preserved assert r3["duration"] == 5 assert r3["transcription_text"] == "hello from yesterday" assert r3["pay_msg_id"] == "pay-abc-123"