[Day8]SQLite FTS5 全文檢索：高亮 + 排名

python

eyeyeyeye 2025-09-28 10:49:20 ‧ 101 瀏覽

分享至

今天把昨天的 crawler.db 升級成全文搜尋，支援：
rebuild：一鍵重建索引（由 links 表建立）
search：關鍵字查詢（FTS5 語法）、排名 bm25、關鍵字高亮 snippet
分頁 --page/--size、CSV 匯出 --out

程式碼（存成 fts_search.py）
把下面整段貼到 project/fts_search.py。

# fts_search.py — Day 8：SQLite FTS5 全文檢索（高亮＋排名）
import sqlite3, argparse, csv, math, os, sys

def open_db(dbname: str):
    if not os.path.exists(dbname):
        sys.exit(f"❌ 找不到資料庫：{dbname}，請先完成 Day 6 匯入。")
    con = sqlite3.connect(dbname)
    con.row_factory = sqlite3.Row
    return con

def ensure_links_table(con: sqlite3.Connection):
    cur = con.cursor()
    cur.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='links'")
    if not cur.fetchone():
        sys.exit("❌ 找不到 links 資料表，請先跑 Day 6 建表並匯入資料。")

def ensure_fts(con: sqlite3.Connection):
    """建立 FTS5 索引與同步觸發器；若 FTS5 不可用則給出清楚訊息。"""
    cur = con.cursor()
    try:
        # 建立全文索引（外部內容表：links）
        cur.execute("""
            CREATE VIRTUAL TABLE IF NOT EXISTS fts_links
            USING fts5(
                text, url,
                content='links',
                content_rowid='id',
                tokenize='unicode61'
            );
        """)
        # 讓 FTS 跟 content 表同步（新增/刪除/更新）
        cur.executescript("""
            CREATE TRIGGER IF NOT EXISTS links_ai AFTER INSERT ON links BEGIN
              INSERT INTO fts_links(rowid, text, url) VALUES (new.id, new.text, new.url);
            END;
            CREATE TRIGGER IF NOT EXISTS links_ad AFTER DELETE ON links BEGIN
              INSERT INTO fts_links(fts_links, rowid, text, url)
              VALUES('delete', old.id, old.text, old.url);
            END;
            CREATE TRIGGER IF NOT EXISTS links_au AFTER UPDATE ON links BEGIN
              INSERT INTO fts_links(fts_links, rowid, text, url)
              VALUES('delete', old.id, old.text, old.url);
              INSERT INTO fts_links(rowid, text, url) VALUES (new.id, new.text, new.url);
            END;
        """)
        con.commit()
    except sqlite3.OperationalError as e:
        if "no such module: fts5" in str(e).lower():
            sys.exit("❌ 這個 Python 內建的 SQLite 沒啟用 FTS5。請更新 Python（3.11+ 通常內建），或改用支援 FTS5 的環境。")
        raise

def rebuild(con: sqlite3.Connection):
    """從 links 表重建全文索引（最快捷的方式）"""
    ensure_fts(con)
    cur = con.cursor()
    # FTS5 內建的 rebuild 指令：會從 content 表（links）重灌索引
    cur.execute("INSERT INTO fts_links(fts_links) VALUES('rebuild');")
    con.commit()

def search(con: sqlite3.Connection, q: str, page: int, size: int):
    ensure_fts(con)
    offset = (page - 1) * size
    cur = con.cursor()

    # 總筆數（符合 MATCH 的列數）
    cur.execute("SELECT COUNT(*) FROM fts_links WHERE fts_links MATCH ?", (q,))
    total = cur.fetchone()[0]

    # 使用 bm25(fts_links, 權重...)：讓 text 權重 1.0，url 權重 0.2
    sql = """
    SELECT l.id,
           snippet(fts_links, 0, '[', ']', '…', 10) AS stext,
           l.url,
           bm25(fts_links, 1.0, 0.2) AS rank
    FROM fts_links
    JOIN links AS l ON l.id = fts_links.rowid
    WHERE fts_links MATCH ?
    ORDER BY rank ASC
    LIMIT ? OFFSET ?;
    """
    cur.execute(sql, (q, size, offset))
    rows = cur.fetchall()
    return total, rows

def to_csv(rows, out_path: str):
    with open(out_path, "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(["id", "text_highlight", "url", "rank"])
        for r in rows:
            w.writerow([r["id"], r["stext"], r["url"], f'{r["rank"]:.3f}'])
    print(f"✅ 已輸出 CSV：{out_path}（{len(rows)} 筆）")

def main():
    ap = argparse.ArgumentParser(description="Day 8：SQLite FTS5 全文檢索（高亮 + 排名）")
    ap.add_argument("--db", default="crawler.db", help="資料庫檔案（預設 crawler.db）")
    sub = ap.add_subparsers(dest="cmd", required=True)

    p_re = sub.add_parser("rebuild", help="從 links 表重建全文索引")
    p_se = sub.add_parser("search", help="全文檢索")
    p_se.add_argument("--q", required=True, help="FTS5 查詢字串，例如：python OR 開發 或 \"鐵人賽\"")
    p_se.add_argument("--page", type=int, default=1, help="頁碼（從 1 開始）")
    p_se.add_argument("--size", type=int, default=10, help="每頁筆數")
    p_se.add_argument("--out", help="將查詢結果匯出為 CSV")

    args = ap.parse_args()
    con = open_db(args.db)
    ensure_links_table(con)

    if args.cmd == "rebuild":
        rebuild(con)
        print("✅ 已重建全文索引 fts_links。")
    else:
        total, rows = search(con, args.q, args.page, args.size)
        pages = max(1, math.ceil(total / args.size)) if total else 1
        print(f"🔎 查詢：{args.q} | 符合：{total} 筆 | 第 {args.page}/{pages} 頁（每頁 {args.size}）")
        for r in rows:
            print(f"[{r['id']:>4}] {r['stext']}  -> {r['url']}  (rank={r['rank']:.3f})")
        if args.out:
            to_csv(rows, args.out)

    con.close()

if __name__ == "__main__":
    main()

怎麼用

先建立索引（只要跑一次，或資料有大變動再重建）
python fts_search.py rebuild
搜尋（FTS5 查詢語法）

# 關鍵字（OR / AND / 引號短語）
python fts_search.py search --q python
python fts_search.py search --q "鐵人賽"
python fts_search.py search --q "python OR 資料庫"
python fts_search.py search --q "\"資料 科學\""   # 短語查詢

# 分頁 & 匯出
python fts_search.py search --q python --page 2 --size 5 --out fts_page2.csv

實作:

今天學到什麼
CREATE VIRTUAL TABLE ... USING fts5(...) 就能把 SQLite 變成小型全文檢索引擎。
content='links' + 觸發器 → 讓 FTS 與來源表同步；INSERT INTO fts_links(fts_links) VALUES('rebuild') 可一鍵重建索引。
bm25() 提供排名；snippet()/highlight() 可做關鍵字高亮。
若遇到 no such module: fts5：你的 Python/SQLite 沒啟用 FTS5，更新 Python（Windows/Mac 官方版通常內建）即可。