如何用Python读取Fun-ASR数据库？脚本示例分享

如何用Python读取Fun-ASR数据库？脚本示例分享

Fun-ASR作为钉钉与通义实验室联合推出的本地化语音识别系统，其轻量、离线、易部署的特性深受开发者欢迎。但很多用户在使用过程中会忽略一个关键事实：所有识别历史并非临时缓存，而是持久化存储在一个标准SQLite数据库中——webui/data/history.db。

这个文件看似普通，却承载着你每一次语音转写的成果：会议录音的文字稿、客服对话的结构化记录、访谈内容的原始文本……一旦误删或磁盘损坏，这些数据将无法通过WebUI界面恢复。而官方WebUI并未提供导出、筛选或批量分析功能，这就意味着——真正掌控数据的能力，必须由你自己构建。

本文不讲模型原理，不谈部署技巧，只聚焦一个务实问题：如何用Python安全、稳定、可扩展地读取Fun-ASR的history.db数据库？从基础连接到字段解析，从时间处理到结果提取，附带多个开箱即用的脚本示例，帮你把“识别历史”真正变成可查询、可分析、可集成的数据资产。

1. 数据库结构解析：先看懂它长什么样

Fun-ASR使用SQLite作为历史记录的底层存储引擎，这是最符合本地应用特性的选择：零配置、单文件、跨平台、无需服务进程。但它的表结构并非完全公开，需通过命令行快速探查。

1.1 查看真实表结构

在Fun-ASR项目根目录下执行：

sqlite3 webui/data/history.db ".schema"

典型输出如下（v1.0.0版本实测）：

CREATE TABLE recognition_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp INTEGER NOT NULL, filename TEXT NOT NULL, filepath TEXT, language TEXT DEFAULT 'zh', result_text TEXT, itn_text TEXT, hotwords TEXT, itn_enabled BOOLEAN DEFAULT 1, model_name TEXT DEFAULT 'Fun-ASR-Nano-2512', duration_ms INTEGER, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );

关键字段说明：

• timestamp：Unix时间戳（秒级），非ISO格式，需转换
• filename：上传时的原始文件名（如meeting_20250412.mp3）
• filepath：服务端保存的绝对路径（如/home/user/funasr/webui/data/audio/meeting_20250412.mp3）
• result_text：原始识别结果（含口语化表达）
• itn_text：启用ITN后的规整文本（如“二零二五年”→“2025年”）
• hotwords：生效的热词列表（JSON字符串格式，如["开放时间","客服电话"]）

注意：字段名可能随版本微调，首次使用前务必执行.schema确认结构，避免脚本因字段不存在而报错。

1.2 字段类型与业务含义映射

这个映射表是你编写健壮脚本的基础。它提醒你：不是所有字段都必然有值，也不是所有类型都如表面所示。

2. 基础读取脚本：安全连接与结果打印

以下是一个生产环境可用的最小可行脚本，已内置错误处理、字段容错和时间格式化逻辑：

# read_history_basic.py import sqlite3 import json from datetime import datetime from pathlib import Path def read_all_records(db_path: str = "webui/data/history.db") -> list: """ 安全读取全部识别历史记录 Args: db_path: SQLite数据库文件路径 Returns: list[dict]: 每条记录为字典，包含标准化字段 """ # 验证数据库文件存在 if not Path(db_path).exists(): raise FileNotFoundError(f"数据库文件未找到: {db_path}") records = [] try: conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row # 启用列名访问 cursor = conn.cursor() # 查询所有字段，按时间倒序（最新在前） cursor.execute(""" SELECT id, timestamp, filename, filepath, language, result_text, itn_text, hotwords, itn_enabled, model_name, duration_ms FROM recognition_history ORDER BY timestamp DESC """) for row in cursor.fetchall(): # 构建标准化记录字典 record = { "id": row["id"], "time": datetime.fromtimestamp(row["timestamp"]).strftime("%Y-%m-%d %H:%M:%S"), "filename": row["filename"].strip() if row["filename"] else "", "filepath": row["filepath"].strip() if row["filepath"] else "", "language": (row["language"] or "zh").lower(), "result": (row["result_text"] or "").strip(), "itn_result": (row["itn_text"] or "").strip(), "hotwords": [], "itn_enabled": bool(row["itn_enabled"]), "model": row["model_name"] or "unknown", "duration_ms": row["duration_ms"] or 0 } # 解析热词（兼容空值和JSON格式） if row["hotwords"]: try: record["hotwords"] = json.loads(row["hotwords"]) except (json.JSONDecodeError, TypeError): record["hotwords"] = [row["hotwords"]] # 当作单字符串处理 records.append(record) except sqlite3.Error as e: print(f"数据库读取失败: {e}") return [] finally: if 'conn' in locals(): conn.close() return records if __name__ == "__main__": # 读取并打印前5条记录 history = read_all_records() print(f"共读取 {len(history)} 条历史记录\n") for i, rec in enumerate(history[:5], 1): print(f"--- 第 {i} 条记录 ---") print(f"ID: {rec['id']}") print(f"时间: {rec['time']}") print(f"文件: {rec['filename']}") print(f"语言: {rec['language']}") print(f"模型: {rec['model']}") print(f"原始结果: {rec['result'][:60]}{'...' if len(rec['result']) > 60 else ''}") print(f"规整结果: {rec['itn_result'][:60]}{'...' if len(rec['itn_result']) > 60 else ''}") print(f"热词: {rec['hotwords']}") print()

脚本亮点：

• 使用sqlite3.Row实现字段名访问，避免位置索引错误
• 对所有TEXT字段做.strip()和空值检查，防止None引发异常
• hotwords字段兼容JSON数组和纯字符串两种旧版格式
• 时间戳自动转为可读格式，无需手动计算
• 全流程异常捕获，数据库损坏时不会中断主程序

运行后，你将看到清晰的结构化输出，每条记录都是一个标准字典，可直接用于后续分析。

3. 进阶实用脚本：按需筛选与导出

基础读取只是起点。在实际工作中，你往往需要：按日期范围提取会议纪要、按语言筛选英文访谈、导出为CSV供Excel分析、或生成统计报告。以下三个脚本覆盖高频场景。

3.1 按日期范围导出记录（export_by_date.py）

# export_by_date.py import sqlite3 import json from datetime import datetime, timedelta import csv from pathlib import Path def export_records_by_date( db_path: str = "webui/data/history.db", start_date: str = None, end_date: str = None, output_csv: str = "funasr_export.csv" ): """ 按日期范围导出识别记录到CSV Args: db_path: 数据库路径 start_date: 开始日期，格式 "2025-04-01" end_date: 结束日期，格式 "2025-04-30" output_csv: 输出CSV文件名 """ # 计算时间戳范围 if not start_date: start_ts = 0 else: start_dt = datetime.strptime(start_date, "%Y-%m-%d") start_ts = int(start_dt.timestamp()) if not end_date: end_dt = datetime.now() end_ts = int(end_dt.timestamp()) else: end_dt = datetime.strptime(end_date, "%Y-%m-%d") + timedelta(days=1) - timedelta(seconds=1) end_ts = int(end_dt.timestamp()) try: conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute(""" SELECT id, timestamp, filename, language, itn_text, result_text, hotwords, duration_ms FROM recognition_history WHERE timestamp BETWEEN ? AND ? ORDER BY timestamp DESC """, (start_ts, end_ts)) rows = cursor.fetchall() if not rows: print("未找到符合条件的记录") return # 写入CSV with open(output_csv, "w", newline="", encoding="utf-8-sig") as f: writer = csv.writer(f) # 表头 writer.writerow([ "ID", "日期", "时间", "文件名", "语言", "规整文本", "原始文本", "热词", "时长(ms)" ]) for row in rows: ts = datetime.fromtimestamp(row[1]) hotwords = "" if row[6]: try: hw_list = json.loads(row[6]) hotwords = "、".join(hw_list) except: hotwords = str(row[6]) writer.writerow([ row[0], ts.strftime("%Y-%m-%d"), ts.strftime("%H:%M:%S"), row[2], row[3], row[4] or "", row[5] or "", hotwords, row[7] or 0 ]) print(f" 已导出 {len(rows)} 条记录至 {output_csv}") except Exception as e: print(f" 导出失败: {e}") finally: if 'conn' in locals(): conn.close() if __name__ == "__main__": # 示例：导出2025年4月1日至4月15日的所有记录 export_records_by_date( start_date="2025-04-01", end_date="2025-04-15", output_csv="meeting_april_2025.csv" )

使用提示：

• utf-8-sig编码确保Excel能正确识别中文
• end_date自动扩展至当日23:59:59，避免漏掉当天最后一条
• 热词字段自动合并为中文顿号分隔，便于人工阅读

3.2 语言统计与TOP10文件分析（stats_analyzer.py）

# stats_analyzer.py import sqlite3 from collections import Counter, defaultdict from pathlib import Path def analyze_history_stats(db_path: str = "webui/data/history.db"): """ 生成识别历史统计报告 """ try: conn = sqlite3.connect(db_path) cursor = conn.cursor() # 总体统计 cursor.execute("SELECT COUNT(*) FROM recognition_history") total = cursor.fetchone()[0] # 按语言统计 cursor.execute("SELECT language, COUNT(*) FROM recognition_history GROUP BY language") lang_stats = dict(cursor.fetchall()) # 按文件名统计（TOP10高频文件） cursor.execute(""" SELECT filename, COUNT(*) as cnt FROM recognition_history GROUP BY filename ORDER BY cnt DESC LIMIT 10 """) top_files = cursor.fetchall() # 按模型统计 cursor.execute("SELECT model_name, COUNT(*) FROM recognition_history GROUP BY model_name") model_stats = dict(cursor.fetchall()) # 平均时长 cursor.execute("SELECT AVG(duration_ms) FROM recognition_history WHERE duration_ms > 0") avg_duration = cursor.fetchone()[0] # 打印报告 print("=== Fun-ASR 识别历史统计报告 ===\n") print(f" 总记录数: {total}") print(f"\n 语言分布:") for lang, count in lang_stats.items(): percentage = (count / total * 100) if total else 0 print(f" • {lang.upper()}: {count} 条 ({percentage:.1f}%)") print(f"\n TOP10 高频识别文件:") for i, (fname, cnt) in enumerate(top_files, 1): print(f" {i}. {fname} — {cnt} 次") print(f"\n 模型使用情况:") for model, count in model_stats.items(): print(f" • {model}: {count} 次") if avg_duration: print(f"\n⏱ 平均音频时长: {avg_duration/1000:.1f} 秒") # 检查异常：无ITN结果的记录 cursor.execute("SELECT COUNT(*) FROM recognition_history WHERE itn_text IS NULL OR itn_text = ''") no_itn = cursor.fetchone()[0] if no_itn > 0: print(f"\n 提示: {no_itn} 条记录未生成ITN规整文本（可能ITN未启用）") except Exception as e: print(f" 统计分析失败: {e}") finally: if 'conn' in locals(): conn.close() if __name__ == "__main__": analyze_history_stats()

输出示例：

=== Fun-ASR 识别历史统计报告 === 总记录数: 247 语言分布: • ZH: 221 条 (89.5%) • EN: 24 条 (9.7%) • JA: 2 条 (0.8%) TOP10 高频识别文件: 1. weekly_meeting.mp3 — 12 次 2. customer_call_0412.wav — 8 次

3.3 批量提取规整文本到独立文件（extract_itexn_texts.py）

# extract_itn_texts.py import sqlite3 import os from pathlib import Path def extract_itn_texts_to_files( db_path: str = "webui/data/history.db", output_dir: str = "itn_exports" ): """ 将所有ITN规整文本导出为独立文本文件，文件名含时间戳和ID Args: db_path: 数据库路径 output_dir: 输出目录名 """ Path(output_dir).mkdir(exist_ok=True) try: conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute(""" SELECT id, timestamp, filename, itn_text FROM recognition_history WHERE itn_text IS NOT NULL AND itn_text != '' ORDER BY timestamp DESC """) records = cursor.fetchall() if not records: print("未找到有效的ITN文本记录") return for rec in records: rec_id, ts, fname, itn_text = rec # 构建文件名：YYYYMMDD_HHMMSS_ID_原始文件名.txt dt = datetime.fromtimestamp(ts) safe_fname = "".join(c for c in fname if c.isalnum() or c in "._- ") filename = f"{dt.strftime('%Y%m%d_%H%M%S')}_{rec_id}_{safe_fname}.txt" # 替换非法字符 filename = filename.replace(" ", "_").replace("/", "_") # 写入文件 filepath = Path(output_dir) / filename with open(filepath, "w", encoding="utf-8") as f: f.write(f"=== Fun-ASR 识别记录 ID:{rec_id} ===\n") f.write(f"时间: {dt.strftime('%Y-%m-%d %H:%M:%S')}\n") f.write(f"源文件: {fname}\n") f.write(f"语言: zh\n\n") f.write(itn_text.strip()) print(f" 已导出 {len(records)} 份ITN文本至 {output_dir}/") except Exception as e: print(f" 文本导出失败: {e}") finally: if 'conn' in locals(): conn.close() if __name__ == "__main__": extract_itn_texts_to_files()

🗂 生成效果：

itn_exports/ ├── 20250412_143022_156_weekly_meeting_mp3.txt ├── 20250412_101545_155_customer_call_0412_wav.txt └── ...

每个文件都是纯文本，可直接导入Notion、Obsidian或发送给同事审阅。

4. 工程化建议：让脚本真正融入工作流

以上脚本已具备生产可用性，但要让它长期稳定服务于你的团队，还需考虑以下工程实践：

4.1 路径管理：避免硬编码

将数据库路径抽象为配置项，支持环境变量或配置文件：

# config.py import os from pathlib import Path # 方式1：从环境变量读取 DB_PATH = os.getenv("FUNASR_DB_PATH", "webui/data/history.db") # 方式2：从配置文件读取（推荐） CONFIG_FILE = Path("config.yaml") if CONFIG_FILE.exists(): import yaml with open(CONFIG_FILE) as f: cfg = yaml.safe_load(f) DB_PATH = cfg.get("database_path", DB_PATH)

4.2 错误重试与日志记录

对关键操作添加重试机制和日志：

import logging from tenacity import retry, stop_after_attempt, wait_fixed logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", handlers=[logging.FileHandler("funasr_reader.log"), logging.StreamHandler()] ) @retry(stop=stop_after_attempt(3), wait=wait_fixed(2)) def safe_read_db(db_path): try: return read_all_records(db_path) except Exception as e: logging.error(f"数据库读取失败，正在重试... {e}") raise

4.3 与Fun-ASR服务协同

在脚本中加入服务状态检查，避免读取时数据库被占用：

import psutil import os def is_funasr_running(): """检查Fun-ASR WebUI进程是否在运行""" for proc in psutil.process_iter(['pid', 'name', 'cmdline']): try: if "python" in proc.info['name'].lower(): cmdline = proc.info['cmdline'] if cmdline and any("gradio" in c.lower() or "start_app.sh" in c for c in cmdline): return True except (psutil.NoSuchProcess, psutil.AccessDenied): pass return False if is_funasr_running(): print(" Fun-ASR 正在运行，建议停止服务后再执行备份") # 可选：自动提示用户

5. 总结：从数据使用者到数据管理者

Fun-ASR的history.db不是一个黑盒，而是一份结构清晰、可编程访问的本地数据资产。通过本文提供的脚本和方法，你已经掌握了：

• 读懂它：用.schema命令快速掌握表结构，理解每个字段的业务含义；
• 读取它：健壮的Python脚本，自动处理空值、类型转换和异常；
• 筛选它：按日期、语言、文件名等维度精准提取目标数据；
• 导出它：生成CSV、独立文本文件，无缝对接Excel、Notion等工具；
• 分析它：生成统计报告，发现使用规律，优化识别策略。

更重要的是，这些能力让你摆脱了WebUI界面的限制。当业务需要将识别结果同步到CRM、生成日报、或训练领域模型时，你不再需要手动复制粘贴——只需修改几行脚本，数据便自动流转。

最后提醒：

• 备份永远比修复重要：将read_history_basic.py加入每日定时任务，自动保存JSON快照；
• 验证永远比假设可靠：每次升级Fun-ASR后，先运行.schema确认结构未变；
• 自动化永远比手工高效：把export_by_date.py封装成Docker镜像，一键导出全公司会议纪要。

数据的价值，不在于它被存储在哪里，而在于你能否随时、准确、低成本地把它变成行动依据。现在，你已经拥有了这把钥匙。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。