rssKeeper/backend/fulltext_search.py

"""SQLite FTS5 全文搜索封装"""
from sqlalchemy import text
from database import engine


def search_articles(query: str, limit: int = 50, offset: int = 0):
    """全文搜索文章
    返回 [(article_id, title, content_snippet, rank), ...]
    """
    if not query or not query.strip():
        return [], 0

    # 转义 FTS5 特殊字符（双引号、* 等）
    # 简单策略：将用户查询视为一个整体短语，加引号包裹
    query = query.replace('"', '""').strip()
    if not query:
        return [], 0
    # 用双引号包裹，避免 FTS5 布尔操作符被误解析
    query = f'"{query}"'

    conn = engine.raw_connection()
    cursor = conn.cursor()

    try:
        # 使用 FTS5 查询
        sql = """
            SELECT a.id, a.title, a.summary, a.link, a.published_at, a.created_at,
                   f.id as feed_id, f.title as feed_title, f.category,
                   rank
            FROM articles_fts
            JOIN articles a ON articles_fts.rowid = a.id
            JOIN feeds f ON a.feed_id = f.id
            WHERE articles_fts MATCH ?
            ORDER BY rank
            LIMIT ? OFFSET ?
        """
        cursor.execute(sql, (query, limit, offset))
        rows = cursor.fetchall()

        # 获取总数
        count_sql = """
            SELECT COUNT(*) FROM articles_fts WHERE articles_fts MATCH ?
        """
        cursor.execute(count_sql, (query,))
        total = cursor.fetchone()[0]

        results = []
        for row in rows:
            results.append({
                "id": row[0],
                "title": row[1],
                "summary": row[2],
                "link": row[3],
                "published_at": row[4],
                "created_at": row[5],
                "feed_id": row[6],
                "feed_title": row[7],
                "category": row[8],
            })

        return results, total
    except Exception as e:
        # FTS5 查询失败时返回空结果
        return [], 0
    finally:
        cursor.close()
        conn.close()


def rebuild_fts_index():
    """重建 FTS5 索引（数据不一致时使用）"""
    conn = engine.raw_connection()
    cursor = conn.cursor()
    try:
        cursor.execute("DELETE FROM articles_fts")
        cursor.execute("""
            INSERT INTO articles_fts(rowid, title, content)
            SELECT id, title, content FROM articles
        """)
        conn.commit()
        return True
    except Exception:
        return False
    finally:
        cursor.close()
        conn.close()