diff --git a/yunke-server/app/api/routers/search.py b/yunke-server/app/api/routers/search.py index f0f2485..fcf6204 100644 --- a/yunke-server/app/api/routers/search.py +++ b/yunke-server/app/api/routers/search.py @@ -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) diff --git a/yunke-server/app/db/fts.py b/yunke-server/app/db/fts.py new file mode 100644 index 0000000..87bb91d --- /dev/null +++ b/yunke-server/app/db/fts.py @@ -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() diff --git a/yunke-server/tests/test_api_search.py b/yunke-server/tests/test_api_search.py index 8cfce02..b66e6ba 100644 --- a/yunke-server/tests/test_api_search.py +++ b/yunke-server/tests/test_api_search.py @@ -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 diff --git a/yunke-server/tests/test_db_fts.py b/yunke-server/tests/test_db_fts.py new file mode 100644 index 0000000..66c9d50 --- /dev/null +++ b/yunke-server/tests/test_db_fts.py @@ -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