企业官网建设流程全解析

富阳网站建设,给文字做网站链接,网站平台做期货,wordpress采集素材教程如何用Python读取Fun-ASR数据库#xff1f;脚本示例分享 Fun-ASR作为钉钉与通义实验室联合推出的本地化语音识别系统#xff0c;其轻量、离线、易部署的特性深受开发者欢迎。但很多用户在使用过程中会忽略一个关键事实#xff1a;所有识别历史并非临时缓存#xff0c;而是…如何用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 );关键字段说明timestampUnix时间戳秒级非ISO格式需转换filename上传时的原始文件名如meeting_20250412.mp3filepath服务端保存的绝对路径如/home/user/funasr/webui/data/audio/meeting_20250412.mp3result_text原始识别结果含口语化表达itn_text启用ITN后的规整文本如“二零二五年”→“2025年”hotwords生效的热词列表JSON字符串格式如[开放时间,客服电话]注意字段名可能随版本微调首次使用前务必执行.schema确认结构避免脚本因字段不存在而报错。1.2 字段类型与业务含义映射字段名类型是否为空实际用途Python处理建议idINTEGERNOT NULL唯一任务ID直接转为inttimestampINTEGERNOT NULL识别发起时间戳datetime.fromtimestamp()转换filenameTEXTNOT NULL用户可见文件名保留原字符串用于归档命名filepathTEXTNULL服务端存储路径检查是否为空避免os.path异常languageTEXTDEFAULT zh识别语言代码统一转小写支持zh/en/ja判断result_textTEXTNULL原始ASR输出.strip()去首尾空格防空值itn_textTEXTNULLITN规整后文本优先使用此字段更符合书面阅读习惯hotwordsTEXTNULL热词列表JSONjson.loads()解析为listitn_enabledBOOLEANDEFAULT 1ITN开关状态转为bool()注意SQLite无原生布尔类型这个映射表是你编写健壮脚本的基础。它提醒你不是所有字段都必然有值也不是所有类型都如表面所示。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(fID: {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(days1) - timedelta(seconds1) 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, encodingutf-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_date2025-04-01, end_date2025-04-15, output_csvmeeting_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_okTrue) 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, encodingutf-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( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[logging.FileHandler(funasr_reader.log), logging.StreamHandler()] ) retry(stopstop_after_attempt(3), waitwait_fixed(2)) def safe_read_db(db_path): try: return read_all_records(db_path) except Exception as e: logging.error(f数据库读取失败正在重试... {e}) raise4.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星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。