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

fix: split message type into normalized label + raw code

Browse Source 
The scraper was writing a normalized type label ("text"/"unknown"/...) into
Message.type, but the column was Integer, so the storage layer coerced it to
NULL and the raw code was only recoverable from raw_json. Align with design
§4.3, which specifies two fields: `type` (normalized enum) and `type_raw`
(original code).

- Model (Message): type -> String(32) normalized label; add type_raw Integer
  for the raw upstream type code.
- Scraper: _normalize_message sets both `type` (label) and `type_raw`
  (coerced int from messageType/type); _is_divider now checks raw -2.
- Repos: type_raw joins _INT_FIELDS; Message.type stored as-is (string).
- Migration 96975050fa79: add type_raw + alter type Integer->String(32),
  wrapped in batch_alter_table so it runs on SQLite and MySQL.
- env.py: enable render_as_batch on SQLite (dev/test) for portable batch
  migrations; MySQL still uses direct ALTER.
- Tests updated end-to-end (scraper, repos, models).

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
浅念
2026-08-01 12:04:08 +08:00
parent 854efd3119
commit 5a2ff58af3
8 changed files with 176 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yunke-server/app/db/models.py
+6 -1
View File
@@ -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)
yunke-server/app/scraper/wechat.py
+44 -7
View File
@@ -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
)
yunke-server/app/storage/repos.py
+9 -9
View File
@@ -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``
yunke-server/migrations/env.py
+9
View File
@@ -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():
yunke-server/migrations/versions/96975050fa79_message_type_split_normalized_label_and_.py
+59
View File
@@ -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 ###
yunke-server/tests/test_models.py
+4 -1
View File
@@ -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()
yunke-server/tests/test_repos.py
+41 -10
View File
@@ -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()
yunke-server/tests/test_scrape_wechat_messages.py
+4 -1
View File
@@ -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"