はじめに
動画を文字起こしすると、1本ずつ読む作業は楽になります。しかし、講義、会議、インタビューが数十本に増えると、別の問題が出てきます。
「インデックスの更新について話していたのは、どの動画の何分だったか」
ファイルごとにテキスト検索しても見つけられますが、動画数が増えるほど手間がかかります。そこで本記事では、タイムスタンプ付き文字起こしをSQLiteへ格納し、複数動画を横断検索できる小さなCLIを作ります。
検索結果は本文だけではなく、元動画、話者、再生時刻を返します。さらに、結果IDを指定するとffplayで該当箇所の5秒前から再生します。
$ python transcript_index.py search "インデックスの更新"
[1] course_01 00:42:18.000 Speaker 2
[インデックスの更新]方法を説明します
$ python transcript_index.py play 1
# course_01.mp4 の 00:42:13 から20秒間再生
クラウドのベクトルDBやLLM APIは使いません。完全一致に近いキーワード検索なら、単一ファイルで動くSQLiteで十分です。
なぜ日本語検索にtrigramを使うのか
SQLiteのFTS5は全文検索用の仮想テーブルです。ただし、標準のunicode61 tokenizerは空白などを単語境界として扱うため、英語のように空白で単語を区切らない日本語では、期待した部分一致にならないことがあります。
trigram tokenizerは文字列を3文字ずつ重ねて分割します。
インデックス更新
↓
インデ / ンデッ / デック / ックス / クス更 / ス更新
これにより、「インデックス更新」のような日本語フレーズを、形態素解析器なしで検索できます。一方、2文字以下の検索語はtrigramを作れません。本記事のCLIでは、3文字未満だけinstr()による全件走査へフォールバックします。
この設計は次のように使い分けます。
| 検索方法 | 向いている用途 | 弱点 |
|---|---|---|
| FTS5 trigram | 製品名、固有名詞、発言の一部 | 同義語は拾わない |
instr() |
1〜2文字の短い語 | データ増加時に遅い |
| ベクトル検索 | 意味が近い発言 | 根拠確認と運用が複雑 |
最初からベクトル検索へ進むのではなく、「正確な語を探したいのか」「意味が近い箇所を探したいのか」を先に分けるのが重要です。
入力データの契約を決める
文字起こしサービスごとの出力形式を、そのまま検索処理へ結合すると、サービスや出力形式を変更するたびにコード全体が壊れます。そこで、検索側では次のJSONLだけを受け付けます。
{"video_id":"course_01","path":"course_01.mp4","start_ms":2538000,"end_ms":2546200,"speaker":"Speaker 2","text":"インデックスの更新方法を説明します"}
{"video_id":"course_01","path":"course_01.mp4","start_ms":2546200,"end_ms":2554000,"speaker":"Speaker 1","text":"検索結果から元動画へ戻れるようにします"}
1行が1発話区間です。
-
video_id: 動画を識別する安定したID -
path: 元動画へのパス -
start_ms/end_ms: 動画先頭を0としたミリ秒 -
speaker: 話者ラベル。人物の実名とは限らない -
text: 発話本文
重要なのは、要約文ではなく発話単位の原文を保存することです。要約だけを検索対象にすると、検索結果から元発言へ戻れません。
元データの作成には、タイムスタンプと話者情報を取得できる文字起こし手段を使います。たとえば最大10GBの動画をタイムスタンプ付きで処理できるVideo Transcriberのデスクトップ版で長時間動画を処理し、得られた結果を上記の中間形式へ正規化できます。ここで示すJSONLは本記事独自の中間形式であり、製品固有のAPIレスポンスを表したものではありません。
実装するCLIの構成
ディレクトリは次の形にします。
transcript-search/
├─ transcript_index.py
├─ transcripts.db
└─ data/
├─ course_01.jsonl
└─ meeting_07.jsonl
必要なものはPython 3.11以降です。検索だけなら外部ライブラリは不要です。動画へジャンプする機能には、FFmpegに含まれるffplayを使います。
FTS5とtrigramの利用可否を実行時に確認する
Pythonのsqlite3が使用するSQLiteは、環境によってビルドオプションとバージョンが異なります。バージョン文字列だけで決めつけず、実際に一時テーブルを作って確認します。
def require_trigram(conn: sqlite3.Connection) -> None:
try:
conn.execute(
"CREATE VIRTUAL TABLE temp.fts_probe "
"USING fts5(text, tokenize='trigram')"
)
conn.execute("DROP TABLE temp.fts_probe")
except sqlite3.OperationalError as exc:
raise SystemExit(
f"FTS5 trigram is unavailable in SQLite "
f"{sqlite3.sqlite_version}: {exc}"
) from exc
PRAGMA compile_optionsを見る方法もありますが、最終的に必要なのは「この接続でtrigram仮想テーブルを作れるか」です。そのため、機能そのものを試す方が確実です。
インデックスを構築する
今回は検索用DBを元データから再生成する運用にします。更新頻度が低い個人用アーカイブなら、複雑な同期処理を入れるより壊れにくくなります。
以下がインデックス作成部分です。
import json
import sqlite3
from pathlib import Path
DB_PATH = Path("transcripts.db")
SCHEMA = """
CREATE VIRTUAL TABLE segments_fts USING fts5(
video_id UNINDEXED,
path UNINDEXED,
start_ms UNINDEXED,
end_ms UNINDEXED,
speaker,
text,
tokenize='trigram'
);
"""
def build_index(root: Path) -> None:
conn = sqlite3.connect(DB_PATH)
require_trigram(conn)
# transcripts.dbは再生成可能な専用インデックスとして扱う
conn.execute("DROP TABLE IF EXISTS segments_fts")
conn.executescript(SCHEMA)
required = {"video_id", "path", "start_ms", "end_ms", "text"}
count = 0
for jsonl_path in sorted(root.rglob("*.jsonl")):
with jsonl_path.open(encoding="utf-8") as source:
for line_no, line in enumerate(source, 1):
if not line.strip():
continue
row = json.loads(line)
missing = required - row.keys()
if missing:
raise ValueError(
f"{jsonl_path}:{line_no}: missing {sorted(missing)}"
)
start_ms = int(row["start_ms"])
end_ms = int(row["end_ms"])
text = str(row["text"]).strip()
if start_ms < 0 or end_ms <= start_ms or not text:
raise ValueError(
f"{jsonl_path}:{line_no}: invalid segment"
)
media_path = Path(row["path"]).expanduser()
if not media_path.is_absolute():
media_path = (jsonl_path.parent / media_path).resolve()
conn.execute(
"""
INSERT INTO segments_fts(
video_id, path, start_ms, end_ms, speaker, text
) VALUES (?, ?, ?, ?, ?, ?)
""",
(
str(row["video_id"]),
str(media_path),
start_ms,
end_ms,
str(row.get("speaker", "")),
text,
),
)
count += 1
conn.execute("INSERT INTO segments_fts(segments_fts) VALUES ('optimize')")
conn.commit()
conn.close()
print(f"indexed {count} segments")
UNINDEXEDを付けた列は、検索対象にはしないものの結果表示に必要なメタデータです。speakerとtextだけがインデックス化されます。
この例ではDROP TABLEを使うため、transcripts.dbは必ず再生成可能な専用DBにしてください。既存業務DBへコードをそのまま適用してはいけません。
検索結果へ元動画の位置を残す
FTS5の検索構文として解釈させたいわけではないので、ユーザー入力は二重引用符で囲み、引用符自体もエスケープします。
def quote_fts_phrase(value: str) -> str:
return '"' + value.replace('"', '""') + '"'
def format_ms(value: int) -> str:
hours, remain = divmod(value, 3_600_000)
minutes, remain = divmod(remain, 60_000)
seconds, millis = divmod(remain, 1_000)
return f"{hours:02d}:{minutes:02d}:{seconds:02d}.{millis:03d}"
def search(query: str, limit: int = 20) -> None:
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
if len(query) < 3:
rows = conn.execute(
"""
SELECT rowid, * FROM segments_fts
WHERE instr(text, ?) > 0 OR instr(speaker, ?) > 0
ORDER BY video_id, CAST(start_ms AS INTEGER)
LIMIT ?
""",
(query, query, limit),
).fetchall()
else:
rows = conn.execute(
"""
SELECT rowid, video_id, path, start_ms, end_ms, speaker,
snippet(segments_fts, 5, '[', ']', '…', 24) AS text
FROM segments_fts
WHERE segments_fts MATCH ?
ORDER BY rank
LIMIT ?
""",
(quote_fts_phrase(query), limit),
).fetchall()
for row in rows:
print(
f"[{row['rowid']}] {row['video_id']} "
f"{format_ms(int(row['start_ms']))} {row['speaker'] or '-'}\n"
f" {row['text']}"
)
if not rows:
print("no matches")
conn.close()
snippet()の第2引数5は、6番目の列であるtextを意味します。ヒット部分を[と]で囲むため、長い発話でも該当位置を見つけやすくなります。
ORDER BY rankは関連度順です。発言時刻順に読みたい場合は、取得後にvideo_idとstart_msで並べ替える方が分かりやすい場合もあります。
結果IDから元動画を開く
検索結果のrowidを使い、元動画を該当箇所の5秒前から20秒間だけ再生します。
import subprocess
def play(segment_id: int) -> None:
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
row = conn.execute(
"SELECT path, start_ms FROM segments_fts WHERE rowid = ?",
(segment_id,),
).fetchone()
conn.close()
if row is None:
raise SystemExit(f"segment not found: {segment_id}")
media_path = Path(row["path"])
if not media_path.exists():
raise SystemExit(f"media file not found: {media_path}")
start_sec = max(0.0, int(row["start_ms"]) / 1000 - 5)
subprocess.run(
[
"ffplay", "-loglevel", "warning",
"-ss", str(start_sec), "-t", "20",
"-autoexit", str(media_path),
],
check=False,
)
5秒前から再生するのは、検索語を含む一文だけでなく、質問や直前の条件を確認するためです。文字起こしは検索用インデックスとして使い、引用や決定事項は元音声へ戻って確認します。
CLIとしてまとめる
最後にargparseから各関数を呼びます。
import argparse
def main() -> None:
parser = argparse.ArgumentParser()
sub = parser.add_subparsers(dest="command", required=True)
build_parser = sub.add_parser("build")
build_parser.add_argument("root", type=Path)
search_parser = sub.add_parser("search")
search_parser.add_argument("query")
search_parser.add_argument("--limit", type=int, default=20)
play_parser = sub.add_parser("play")
play_parser.add_argument("segment_id", type=int)
args = parser.parse_args()
if args.command == "build":
build_index(args.root)
elif args.command == "search":
search(args.query, args.limit)
else:
play(args.segment_id)
if __name__ == "__main__":
main()
実行方法は次のとおりです。
python transcript_index.py build ./data
python transcript_index.py search "インデックスの更新"
python transcript_index.py search "予算" --limit 10
python transcript_index.py play 1
運用で追加したいチェック
実データで運用するなら、検索機能より先にデータ契約を守るチェックを追加します。
- 同じ
video_idで異なる動画パスを登録していないか -
start_ms < end_msになっているか - 時刻が元動画の長さを超えていないか
- 同じ区間を重複登録していないか
- 元動画を移動したあと、パスを更新したか
- 話者ラベルを実名だと誤解していないか
また、機密会議の文字起こしでは、検索DBにも元発言が保存されます。単一ファイルだから安全なのではありません。保存場所、バックアップ、アクセス権、削除ルールは元動画と同じ基準で管理します。
まとめ
タイムスタンプ付き文字起こしは、読むための文書であると同時に、動画へ戻るためのインデックスです。
SQLite FTS5とtrigramを使えば、追加のサーバーを用意せず、日本語の発言を複数動画から横断検索できます。検索結果にvideo_id、時刻、話者、元パスを残せば、「見つける」と「原音で確かめる」を一つの流れにできます。
意味検索が必要になった時点でベクトル検索を追加すればよく、最初の一歩としては、根拠が見えるキーワード検索の方が運用しやすいはずです。