#!/usr/bin/env python3 """ 将CECD2024的cecd2024.db转换为EasyDict的media.db 先用mdict将mdd转换为db(sqlite格式) mdict -x cecd2024.mdd --exdb cecd2024.db结构: CREATE TABLE mdd (entry TEXT NOT NULL, file BLOB NOT NULL); EasyDict media.db结构: CREATE TABLE audios (name TEXT PRIMARY KEY, blob BLOB NOT NULL); CREATE TABLE images (name TEXT PRIMARY KEY, blob BLOB NOT NULL); 用法: python convert_cecd_db_to_media_db.py cecd2024/cecd2024.db -o cecd2024_dict """ import sqlite3 import argparse from pathlib import Path def get_file_extension(filename: str) -> str: """获取文件扩展名（小写）""" return Path(filename).suffix.lower() def is_audio_file(filename: str) -> bool: """判断是否为音频文件""" audio_extensions = {'.mp3', '.wav', '.ogg', '.opus', '.m4a', '.aac', '.flac'} return get_file_extension(filename) in audio_extensions def is_image_file(filename: str) -> bool: """判断是否为图片文件""" image_extensions = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.svg', '.webp', '.ico'} return get_file_extension(filename) in image_extensions def extract_filename(entry_path: str) -> str: """ 从entry路径中提取文件名 例如: \\uk_pron\\a\\agr\\agrar\\agrarian_revolution.mp3 -> agrarian_revolution.mp3 """ # 清理路径分隔符 entry_path = entry_path.replace('\\', '/').strip('/') # 提取文件名 if '/' in entry_path: filename = entry_path.split('/')[-1] else: filename = entry_path return filename def convert_db(source_db: str, output_dir: str, page_size_kb: int = 64): """ 转换cecd2024.db到media.db Args: source_db: 源数据库路径 (cecd2024.db) output_dir: 输出目录 page_size_kb: SQLite页面大小（KB） """ source_db = Path(source_db) output_dir = Path(output_dir) if not source_db.exists(): print(f"错误: 源数据库不存在: {source_db}") return False # 创建输出目录 output_dir.mkdir(parents=True, exist_ok=True) # 输出数据库路径 output_db = output_dir / "media.db" if output_db.exists(): print(f"警告: {output_db} 已存在，将被覆盖") output_db.unlink() print(f"\n正在转换数据库...") print(f"源: {source_db}") print(f"目标: {output_db}") # 连接源数据库 source_conn = sqlite3.connect(source_db) source_cursor = source_conn.cursor() # 连接目标数据库 target_conn = sqlite3.connect(output_db) target_cursor = target_conn.cursor() try: # 设置页面大小 page_size_bytes = page_size_kb * 1024 target_conn.execute(f"PRAGMA page_size = {page_size_bytes}") # 创建表结构 print("\n创建表结构...") target_cursor.execute(""" CREATE TABLE IF NOT EXISTS audios ( name TEXT PRIMARY KEY, blob BLOB NOT NULL ) """) target_cursor.execute(""" CREATE TABLE IF NOT EXISTS images ( name TEXT PRIMARY KEY, blob BLOB NOT NULL ) """) target_conn.commit() print("✓ 表结构创建完成") # 统计信息 audio_count = 0 image_count = 0 skipped_count = 0 duplicate_count = 0 total_audio_size = 0 total_image_size = 0 # 用于去重 audio_names = set() image_names = set() # 批量插入 audio_batch = [] image_batch = [] batch_size = 1000 # 查询所有记录 print("\n正在提取资源...") source_cursor.execute("SELECT entry, file FROM mdd") for idx, (entry, file_blob) in enumerate(source_cursor, 1): # 提取文件名 filename = extract_filename(entry) # 跳过空文件名或非媒体文件 if not filename: skipped_count += 1 continue # 判断文件类型 if is_audio_file(filename): # 检查重复 if filename in audio_names: duplicate_count += 1 continue audio_names.add(filename) audio_batch.append((filename, file_blob)) audio_count += 1 total_audio_size += len(file_blob) if len(audio_batch) >= batch_size: target_cursor.executemany( "INSERT OR REPLACE INTO audios (name, blob) VALUES (?, ?)", audio_batch ) target_conn.commit() audio_batch = [] elif is_image_file(filename): # 检查重复 if filename in image_names: duplicate_count += 1 continue image_names.add(filename) image_batch.append((filename, file_blob)) image_count += 1 total_image_size += len(file_blob) if len(image_batch) >= batch_size: target_cursor.executemany( "INSERT OR REPLACE INTO images (name, blob) VALUES (?, ?)", image_batch ) target_conn.commit() image_batch = [] else: skipped_count += 1 # 进度显示 if idx % 5000 == 0: print(f" 已处理: {idx:,} 个资源 (音频: {audio_count:,}, 图片: {image_count:,}, 跳过: {skipped_count:,})") # 插入剩余的批次 if audio_batch: target_cursor.executemany( "INSERT OR REPLACE INTO audios (name, blob) VALUES (?, ?)", audio_batch ) if image_batch: target_cursor.executemany( "INSERT OR REPLACE INTO images (name, blob) VALUES (?, ?)", image_batch ) target_conn.commit() # 创建索引 print("\n正在创建索引...") target_cursor.execute("CREATE INDEX IF NOT EXISTS idx_audios_name ON audios(name)") target_cursor.execute("CREATE INDEX IF NOT EXISTS idx_images_name ON images(name)") target_conn.commit() print("✓ 索引创建完成") # Vacuum优化 print("\n正在优化数据库...") size_before = output_db.stat().st_size target_cursor.execute("VACUUM") target_conn.commit() size_after = output_db.stat().st_size # 输出统计信息 print("\n" + "="*70) print("转换完成!") print("="*70) print(f"输出文件: {output_db}") print(f"\n资源统计:") print(f" 音频文件: {audio_count:,} 个 ({total_audio_size / 1024 / 1024:.2f} MB)") print(f" 图片文件: {image_count:,} 个 ({total_image_size / 1024 / 1024:.2f} MB)") print(f" 跳过文件: {skipped_count:,} 个") print(f" 重复文件: {duplicate_count:,} 个") print(f"\n数据库大小:") print(f" 优化前: {size_before / 1024 / 1024:.2f} MB") print(f" 优化后: {size_after / 1024 / 1024:.2f} MB") if size_before > 0: print(f" 压缩率: {(1 - size_after / size_before) * 100:.1f}%") return True except Exception as e: print(f"\n错误: {e}") import traceback traceback.print_exc() return False finally: source_conn.close() target_conn.close() def main(): parser = argparse.ArgumentParser( description="将CECD2024的cecd2024.db转换为EasyDict的media.db", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 示例: # 基本用法 python convert_cecd_db_to_media_db.py cecd2024/cecd2024.db -o cecd2024_dict # 自定义页面大小 python convert_cecd_db_to_media_db.py cecd2024/cecd2024.db -o cecd2024_dict --page-size 128 """ ) parser.add_argument( "source_db", help="源数据库路径 (cecd2024.db)" ) parser.add_argument( "-o", "--output", required=True, help="输出目录路径（将在此目录生成media.db）" ) parser.add_argument( "--page-size", type=int, default=64, metavar="KB", help="SQLite页面大小（KB），默认64" ) args = parser.parse_args() success = convert_db(args.source_db, args.output, args.page_size) if not success: exit(1) if __name__ == "__main__": main()