55 lines
2.3 KiB
Python
55 lines
2.3 KiB
Python
#!/usr/bin/env python3
|
||
"""Snapshot a readable SQLite table to JSONL without modifying its source."""
|
||
import argparse
|
||
import json
|
||
import sqlite3
|
||
from pathlib import Path
|
||
|
||
|
||
def export(source, table, output):
|
||
source, output = Path(source).resolve(), Path(output).resolve()
|
||
if source == output or output.exists():
|
||
raise ValueError('输出文件已存在或与源文件相同，请使用新的输出路径')
|
||
with source.open('rb') as file:
|
||
if file.read(16) != b'SQLite format 3\x00':
|
||
raise ValueError('源文件不是明文 SQLite；此脚本不解密数据库')
|
||
db = sqlite3.connect(source.as_uri() + '?mode=ro', uri=True, timeout=15)
|
||
snapshot = sqlite3.connect(':memory:')
|
||
created = False
|
||
try:
|
||
# Backup API includes committed WAL contents; never copy only a live .db.
|
||
db.backup(snapshot)
|
||
snapshot.row_factory = sqlite3.Row
|
||
names = [r[0] for r in snapshot.execute("SELECT name FROM sqlite_master WHERE type='table'")]
|
||
if table not in names:
|
||
raise ValueError('消息表不存在，可用表：' + ', '.join(names))
|
||
quoted = '"' + table.replace('"', '""') + '"'
|
||
with output.open('x', encoding='utf-8') as file:
|
||
created = True
|
||
try:
|
||
output.chmod(0o600)
|
||
except OSError:
|
||
pass
|
||
for row in snapshot.execute('SELECT * FROM ' + quoted):
|
||
record = dict(row)
|
||
if any(isinstance(v, bytes) for v in record.values()):
|
||
raise ValueError('表中含二进制字段，需先解码/转换；不会把 BLOB 假装成消息正文')
|
||
file.write(json.dumps(record, ensure_ascii=False) + '\n')
|
||
except Exception:
|
||
if created and output.exists():
|
||
output.unlink() # Only the output file exclusively created by this invocation.
|
||
raise
|
||
finally:
|
||
db.close()
|
||
snapshot.close()
|
||
|
||
|
||
if __name__ == '__main__':
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument('source', type=Path)
|
||
parser.add_argument('table')
|
||
parser.add_argument('output', type=Path)
|
||
args = parser.parse_args()
|
||
export(args.source, args.table, args.output)
|
||
print('Readable JSONL written to:', args.output.resolve())
|