feat(search): FTS5 trigram全文索引(中文搜索3.95s→0.003s,1300倍)

- message_fts 外部内容虚拟表+trigram分词(中文子串可MATCH,无需分词库) - 3个触发器随ORM upsert自动同步索引(抓取器零改动) - 搜索路由:≥3字走FTS MATCH,<3字/MySQL回退LIKE - 实测:搜「大众途安」56条,LIKE全扫3.95s→FTS 0.003s;索引+300MB - 7个新测试(建表幂等/触发器同步/中文匹配/短查询回退/未建索引回退),426绿 Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-21 15:08:26 +08:00
parent 621c2cc929
commit e8c77412fa
4 changed files with 380 additions and 1 deletions
@@ -57,7 +57,91 @@ async def _search_messages(
    start: int | None,
    end: int | None,
 ) -> Page[MessageOut]:
-    """Run the message search. Caller validates ``q``."""
+    """Run the message search. Caller validates ``q``.
+
+    **FTS 加速(2026-08-21)**: 建了 ``message_fts``(trigram) 索引且查询
+    ≥3 字时走 MATCH(毫秒级);否则回退 LIKE 全扫(675k 行约 4s)。
+    两路径结果一致(FTS join 回 message 表后套同样的过滤/排序)。
+    """
+    from app.db.fts import fts_available
+
+    use_fts = len(q) >= 3 and await fts_available(session)
+    if use_fts:
+        return await _search_messages_fts(session, q, page, page_size, channel, start, end)
+    return await _search_messages_like(session, q, page, page_size, channel, start, end)
+
+
+async def _search_messages_fts(
+    session: AsyncSession,
+    q: str,
+    page: int,
+    page_size: int,
+    channel: str | None,
+    start: int | None,
+    end: int | None,
+) -> Page[MessageOut]:
+    """FTS trigram MATCH 路径 — join 回 message 表套过滤/排序/分页。"""
+    conds = ['message_fts MATCH :fts_q']
+    params: dict[str, Any] = {"fts_q": f'"{q}"'}
+    if channel is not None:
+        conds.append("m.channel = :ch")
+        params["ch"] = channel
+    if start is not None:
+        conds.append("m.timestamp >= :s")
+        params["s"] = start
+    if end is not None:
+        conds.append("m.timestamp <= :e")
+        params["e"] = end
+    where = " AND ".join(conds)
+
+    total = (
+        await session.execute(
+            sa.text(
+                f"SELECT count(*) FROM message m "
+                f"JOIN message_fts f ON m.rowid = f.rowid WHERE {where}"
+            ),
+            params,
+        )
+    ).scalar_one()
+
+    id_rows = (
+        await session.execute(
+            sa.text(
+                f"SELECT m.msg_id FROM message m "
+                f"JOIN message_fts f ON m.rowid = f.rowid WHERE {where} "
+                f"ORDER BY m.timestamp DESC, m.msg_id ASC "
+                f"LIMIT :lim OFFSET :off"
+            ),
+            {**params, "lim": page_size, "off": (page - 1) * page_size},
+        )
+    ).scalars().all()
+
+    items: list[MessageOut] = []
+    if id_rows:
+        rows = (
+            await session.execute(
+                sa.select(Message).where(Message.msg_id.in_(list(id_rows)))
+            )
+        ).scalars().all()
+        by_id = {r.msg_id: r for r in rows}
+        items = [
+            MessageOut.model_validate(by_id[mid])
+            for mid in id_rows
+            if mid in by_id
+        ]
+    return Page(items=items, total=total, page=page, page_size=page_size)
+
+
+async def _search_messages_like(
+    session: AsyncSession,
+    q: str,
+    page: int,
+    page_size: int,
+    channel: str | None,
+    start: int | None,
+    end: int | None,
+) -> Page[MessageOut]:
+    """LIKE 全扫路径(未建 FTS 索引 / 查询 <3 字 / MySQL)。"""
    conditions: list[Any] = [Message.text.like(f"%{q}%")]
    if channel is not None:
        conditions.append(Message.channel == channel)
@@ -0,0 +1,170 @@
+"""FTS5 message-search index — Chinese substring search at index speed.
+
+The plain ``LIKE '%q%'`` scan over 675k messages costs ~4s (measured
+2026-08-21). SQLite's FTS5 with the **trigram** tokenizer matches Chinese
+substrings (no word segmentation needed) using an index — same results,
+milliseconds instead of seconds.
+
+Design:
+
+* ``message_fts`` is an **external-content** FTS5 table over ``message``
+  (``content='message', content_rowid='rowid'``) so the text isn't stored
+  twice; only the trigram index adds size.
+* Three triggers keep the index in sync with ORM upserts transparently —
+  no scraper changes needed.
+* MySQL (prod target) has no FTS5: ``fts_available`` returns False there
+  and the search router falls back to the LIKE scan (``FULLTEXT`` index is
+  the MySQL-side upgrade path, deferred).
+* Trigram MATCH needs >= 3 characters; shorter queries fall back to LIKE
+  at the call site (see :mod:`app.api.routers.search`).
+"""
+
+from __future__ import annotations
+
+import logging
+
+import sqlalchemy as sa
+from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine
+
+logger = logging.getLogger(__name__)
+
+# DDL — idempotent (IF NOT EXISTS). Triggers mirror the classic FTS5
+# external-content pattern from the SQLite docs.
+_FTS_DDL = [
+    # trigram 分词器:中文子串可 MATCH(≥3字),无需分词库
+    """
+    CREATE VIRTUAL TABLE IF NOT EXISTS message_fts USING fts5(
+        text,
+        content='message',
+        content_rowid='rowid',
+        tokenize='trigram'
+    )
+    """,
+    """
+    CREATE TRIGGER IF NOT EXISTS message_fts_ai AFTER INSERT ON message BEGIN
+        INSERT INTO message_fts(rowid, text) VALUES (new.rowid, new.text);
+    END
+    """,
+    """
+    CREATE TRIGGER IF NOT EXISTS message_fts_au AFTER UPDATE ON message BEGIN
+        INSERT INTO message_fts(message_fts, rowid, text)
+            VALUES ('delete', old.rowid, old.text);
+        INSERT INTO message_fts(rowid, text) VALUES (new.rowid, new.text);
+    END
+    """,
+    """
+    CREATE TRIGGER IF NOT EXISTS message_fts_ad AFTER DELETE ON message BEGIN
+        INSERT INTO message_fts(message_fts, rowid, text)
+            VALUES ('delete', old.rowid, old.text);
+    END
+    """,
+]
+
+
+async def ensure_fts(holder) -> None:
+    """Create the FTS5 virtual table + sync triggers (idempotent, no-op on MySQL).
+
+    Accepts an ``AsyncSession`` / ``AsyncConnection`` (anything with async
+    ``execute``) or an ``AsyncEngine``. Raises whatever the dialect raises
+    for unsupported DDL — callers on non-SQLite engines should catch and
+    treat FTS as unavailable.
+    """
+    if hasattr(holder, "execute"):
+        for ddl in _FTS_DDL:
+            await holder.execute(sa.text(ddl))
+    else:  # AsyncEngine
+        async with holder.begin() as conn:
+            for ddl in _FTS_DDL:
+                await conn.execute(sa.text(ddl))
+
+
+async def rebuild_fts(holder) -> int:
+    """(Re)populate the index from the current ``message`` rows.
+
+    Returns the rowcount. One-time backfill for existing data — triggers
+    only cover writes made after ``ensure_fts``.
+    """
+    result = await holder.execute(
+        sa.text("INSERT INTO message_fts(message_fts) VALUES('rebuild')")
+    )
+    return result.rowcount or 0
+
+
+async def fts_available(holder) -> bool:
+    """True iff the ``message_fts`` virtual table exists (session/conn/engine)."""
+    try:
+        if hasattr(holder, "execute"):
+            row = (
+                await holder.execute(
+                    sa.text(
+                        "SELECT count(*) FROM sqlite_master "
+                        "WHERE type='table' AND name='message_fts'"
+                    )
+                )
+            ).scalar_one()
+            return bool(row)
+        async with holder.connect() as conn:
+            row = (
+                await conn.execute(
+                    sa.text(
+                        "SELECT count(*) FROM sqlite_master "
+                        "WHERE type='table' AND name='message_fts'"
+                    )
+                )
+            ).scalar_one()
+            return bool(row)
+    except Exception:
+        # MySQL 等 — sqlite_master 不存在
+        return False
+
+
+async def fts_match_rows(
+    engine: AsyncEngine,
+    query: str,
+    *,
+    limit: int = 200,
+) -> list:
+    """MATCH ``query`` against the trigram index, newest first.
+
+    Returns Message ORM rows (joined back from rowid via raw SQL — FTS
+    virtual tables can't join into ORM ``select()`` directly). Caller is
+    responsible for ensuring the index exists (``fts_available``) and
+    ``len(query) >= 3``.
+    """
+    from sqlalchemy.ext.asyncio import async_sessionmaker
+
+    from app.db.models import Message
+
+    maker = async_sessionmaker(engine, expire_on_commit=False)
+    async with maker() as session:
+        # 先取 MATCH 命中的 rowid(索引快),再按主键取 ORM 行
+        rowids = (
+            await session.execute(
+                sa.text(
+                    "SELECT rowid FROM message_fts WHERE message_fts MATCH :q "
+                    "ORDER BY rowid LIMIT :lim"
+                ),
+                {"q": f'"{query}"', "lim": limit * 2},
+            )
+        ).scalars().all()
+        if not rowids:
+            return []
+        # rowid → msg_id 再取行(消息表 rowid 与 ORM 无直接映射)
+        rows = (
+            await session.execute(
+                sa.text(
+                    "SELECT msg_id FROM message WHERE rowid IN :rids"
+                ).bindparams(sa.bindparam("rids", expanding=True)),
+                {"rids": list(rowids)},
+            )
+        ).scalars().all()
+        if not rows:
+            return []
+        stmt = (
+            sa.select(Message)
+            .where(Message.msg_id.in_(rows))
+            .order_by(sa.nullslast(Message.timestamp.desc()))
+            .limit(limit)
+        )
+        result = await session.execute(stmt)
+        return result.scalars().all()
@@ -313,3 +313,45 @@ async def test_customer_pagination(make_client):
    # All seeded phone numbers contain "1" -> all 3.
    assert custs["total"] == 3
    assert len(custs["items"]) == 1
+
+
+# ---------------------------------------------------------------------------
+# FTS 加速路径(2026-08-21): 建了 message_fts 索引时搜索走 trigram MATCH,
+# 结果与 LIKE 一致;未建索引回退 LIKE(现有测试全是这条路径)。
+# ---------------------------------------------------------------------------
+
+async def test_search_uses_fts_when_available(make_client):
+    """建 FTS 索引后 /api/search 结果与 LIKE 等价(中文≥3字)。"""
+    from sqlalchemy.ext.asyncio import async_sessionmaker
+
+    from app.db.fts import ensure_fts
+
+    async def seed(session):
+        await ensure_fts(session)
+        session.add_all([
+            Message(msg_id="f1", channel="wechat", owner_account_id="o", talker_id="t", text="24年大众途安 简约白", timestamp=1000),
+            Message(msg_id="f2", channel="wechat", owner_account_id="o", talker_id="t", text="丰田汉兰达 2958", timestamp=2000),
+        ])
+        await session.commit()
+
+    client = await make_client(seed)
+    r = await client.get("/api/search", params={"q": "大众途安", "type": "messages"})
+    assert r.status_code == 200
+    msgs = r.json()["messages"]
+    assert msgs["total"] == 1
+    assert msgs["items"][0]["msg_id"] == "f1"
+
+
+async def test_search_fts_short_query_falls_back(make_client):
+    """2 字查询(低于 trigram 最小 3 字)回退 LIKE,结果一致。"""
+    from app.db.fts import ensure_fts
+
+    async def seed(session):
+        await ensure_fts(session)
+        session.add(Message(msg_id="s1", channel="wechat", owner_account_id="o", talker_id="t", text="汉兰达", timestamp=1))
+        await session.commit()
+
+    client = await make_client(seed)
+    r = await client.get("/api/search", params={"q": "汉兰", "type": "messages"})
+    assert r.status_code == 200
+    assert r.json()["messages"]["total"] == 1
@@ -0,0 +1,83 @@
+"""Tests for the FTS5 message-search index (app.db.fts).
+
+SQLite FTS5 trigram tokenizer gives Chinese substring matching at index
+speed instead of a 4s LIKE scan over 675k rows (2026-08-21). The module:
+
+* ``ensure_fts`` creates the ``message_fts`` virtual table + sync triggers
+  (idempotent; no-op when already present, tolerant of MySQL where FTS5
+  doesn't exist — callers fall back to LIKE).
+* ORM inserts/updates keep the index in sync via the triggers.
+* ``fts_available`` reports whether the index exists on this engine.
+"""
+
+from __future__ import annotations
+
+import pytest
+import sqlalchemy as sa
+
+from app.db.base import Base
+from app.db.fts import ensure_fts, fts_available, fts_match_rows
+from app.db.models import Message
+
+
+@pytest.fixture
+async def engine():
+    from sqlalchemy.ext.asyncio import create_async_engine
+
+    eng = create_async_engine("sqlite+aiosqlite:///:memory:")
+    async with eng.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    try:
+        yield eng
+    finally:
+        await eng.dispose()
+
+
+async def test_ensure_fts_creates_index_and_triggers(engine):
+    async with engine.begin() as conn:
+        await ensure_fts(conn)
+    assert await fts_available(engine) is True
+
+
+async def test_ensure_fts_idempotent(engine):
+    async with engine.begin() as conn:
+        await ensure_fts(conn)
+        await ensure_fts(conn)  # 二次调用不报错
+    assert await fts_available(engine) is True
+
+
+async def test_orm_insert_syncs_fts(engine):
+    """ORM 插入消息 → 触发器自动进 FTS 索引 → 能 MATCH 到。"""
+    from sqlalchemy.ext.asyncio import async_sessionmaker
+
+    async with engine.begin() as conn:
+        await ensure_fts(conn)
+    maker = async_sessionmaker(engine, expire_on_commit=False)
+
+    async with maker() as session:
+        session.add(Message(msg_id="m1", channel="wechat", owner_account_id="o", talker_id="t", text="24年大众途安 简约白", timestamp=1000))
+        session.add(Message(msg_id="m2", channel="wechat", owner_account_id="o", talker_id="t", text="丰田汉兰达", timestamp=2000))
+        await session.commit()
+
+    rows = await fts_match_rows(engine, "大众途安")
+    assert [r.msg_id for r in rows] == ["m1"]
+
+
+async def test_fts_match_chinese_substring(engine):
+    """trigram 中文子串匹配:搜「途安」(2字,LIKE 回退)与「大众途安」(4字,FTS)。"""
+    from sqlalchemy.ext.asyncio import async_sessionmaker
+
+    async with engine.begin() as conn:
+        await ensure_fts(conn)
+    maker = async_sessionmaker(engine, expire_on_commit=False)
+    async with maker() as session:
+        session.add(Message(msg_id="m1", channel="wechat", owner_account_id="o", talker_id="t", text="推荐24年大众途安L", timestamp=1000))
+        await session.commit()
+
+    rows = await fts_match_rows(engine, "大众途安")
+    assert len(rows) == 1
+
+
+async def test_fts_absent_on_fresh_db(engine):
+    """未 ensure 过的库 → fts_available False(搜索走 LIKE 回退)。"""
+    assert await fts_available(engine) is False