ax-backend-v1/export_db_to_jsonl.py

"""
将 SQLite 数据库文件导出为 JSONL 格式

使用方法:
    python export_db_to_jsonl.py <db_file_path> [output_file_path]

示例:
    python export_db_to_jsonl.py tmp/user1/sqlite/doc-abc123.db
    python export_db_to_jsonl.py tmp/user1/sqlite/doc-abc123.db output.jsonl
    python export_db_to_jsonl.py D:\code\ax-backend-v1\tmp\default-user\sqlite\doc-a11d6a3cd242.db output.jsonl
"""

import json
import sqlite3
import sys
from pathlib import Path


def export_db_to_jsonl(db_path: str, output_path: str = None):
    """
    将 SQLite 数据库导出为 JSONL 格式
    
    Args:
        db_path: SQLite 数据库文件路径
        output_path: 输出的 JSONL 文件路径（可选，默认为 db_path.jsonl）
    """
    db_file = Path(db_path)
    
    # 检查数据库文件是否存在
    if not db_file.exists():
        print(f"错误: 数据库文件不存在: {db_path}")
        return False
    
    # 确定输出文件路径
    if output_path is None:
        output_path = db_file.with_suffix('.jsonl')
    else:
        output_path = Path(output_path)
    
    try:
        # 连接数据库
        conn = sqlite3.connect(str(db_file))
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()
        
        # 查询所有 blocks，按 block_order 排序
        cursor.execute("""
            SELECT * FROM document_blocks 
            ORDER BY block_order
        """)
        
        rows = cursor.fetchall()
        
        if not rows:
            print(f"警告: 数据库中没有找到任何 blocks")
            return False
        
        # 写入 JSONL 文件
        with open(output_path, 'w', encoding='utf-8') as f:
            for row in rows:
                # 转换为字典
                block = dict(row)
                
                # 解析 JSON 字段
                if block.get('style'):
                    try:
                        block['style'] = json.loads(block['style'])
                    except (json.JSONDecodeError, TypeError):
                        block['style'] = {}
                
                if block.get('metadata'):
                    try:
                        block['metadata'] = json.loads(block['metadata'])
                    except (json.JSONDecodeError, TypeError):
                        block['metadata'] = {}
                
                # content 可能是 JSON（富文本或表格）
                if block.get('content'):
                    try:
                        block['content'] = json.loads(block['content'])
                    except (json.JSONDecodeError, TypeError):
                        pass  # 保持原字符串
                
                # 写入一行 JSON
                f.write(json.dumps(block, ensure_ascii=False) + '\n')
        
        conn.close()
        
        print(f"✓ 成功导出 {len(rows)} 个 blocks")
        print(f"✓ 输出文件: {output_path}")
        return True
        
    except sqlite3.Error as e:
        print(f"错误: 数据库操作失败: {e}")
        return False
    except Exception as e:
        print(f"错误: {e}")
        return False


def main():
    """主函数"""
    if len(sys.argv) < 2:
        print("用法: python export_db_to_jsonl.py <db_file_path> [output_file_path]")
        print("\n示例:")
        print("  python export_db_to_jsonl.py tmp/user1/sqlite/doc-abc123.db")
        print("  python export_db_to_jsonl.py tmp/user1/sqlite/doc-abc123.db output.jsonl")
        sys.exit(1)
    
    db_path = sys.argv[1]
    output_path = sys.argv[2] if len(sys.argv) > 2 else None
    
    print(f"正在导出数据库: {db_path}")
    success = export_db_to_jsonl(db_path, output_path)
    
    if success:
        print("\n导出完成！")
        sys.exit(0)
    else:
        print("\n导出失败！")
        sys.exit(1)


if __name__ == "__main__":
    main()