FUTURE POLICE技巧：让语音分析结果在MySQL里随时可查-程序员充电站

FUTURE POLICE技巧：让语音分析结果在MySQL里随时可查

1. 语音分析数据管理的痛点与解决方案

语音分析技术正在改变我们处理音频数据的方式，但随之而来的数据管理问题却常常被忽视。想象一下这样的场景：你刚刚用FUTURE POLICE完成了100个客服通话录音的分析，得到了包含情感标签、关键词提取、时间戳对齐的丰富数据。这些数据分散在几十个JSON文件中，当你想回答"上周有多少通话出现了负面情绪"这样的简单问题时，却不得不手动打开每个文件进行统计。

这就是为什么我们需要将FUTURE POLICE的分析结果系统化地存储到MySQL数据库中。数据库不仅能解决数据分散的问题，还能带来以下优势：

即时查询能力：通过SQL语句快速获取特定条件的分析结果
多维分析：轻松实现时间趋势、情感分布、关键词关联等复杂分析
历史追溯：完整记录每次分析任务的元数据和结果
系统集成：为后续的可视化报表或业务系统提供数据支持

2. 数据库设计与表结构规划

2.1 核心数据模型设计

基于FUTURE POLICE的输出特性，我们设计了三层数据结构：

任务层：记录分析任务的基本信息
分段层：存储按时间切分的文本和情感数据
明细层：保存关键词、实体等扩展分析结果

这种设计既保持了数据的结构化，又提供了足够的灵活性来适应未来可能新增的分析维度。

2.2 具体表结构实现

-- 分析任务主表 CREATE TABLE analysis_task ( task_id VARCHAR(64) PRIMARY KEY, audio_file VARCHAR(255) NOT NULL, duration FLOAT COMMENT '音频时长(秒)', model_version VARCHAR(50), status ENUM('pending','processing','success','failed') DEFAULT 'pending', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, completed_at TIMESTAMP NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- 文本分段表 CREATE TABLE transcript_segment ( segment_id BIGINT PRIMARY KEY AUTO_INCREMENT, task_id VARCHAR(64) NOT NULL, segment_no INT NOT NULL, start_time FLOAT NOT NULL, end_time FLOAT NOT NULL, text_content TEXT NOT NULL, sentiment VARCHAR(20), confidence FLOAT, FOREIGN KEY (task_id) REFERENCES analysis_task(task_id) ON DELETE CASCADE, INDEX idx_task_segment (task_id, segment_no), INDEX idx_sentiment (sentiment) ); -- 分析明细表 CREATE TABLE analysis_detail ( detail_id BIGINT PRIMARY KEY AUTO_INCREMENT, task_id VARCHAR(64) NOT NULL, segment_id BIGINT NULL, detail_type VARCHAR(50) NOT NULL, detail_value TEXT NOT NULL, confidence FLOAT, FOREIGN KEY (task_id) REFERENCES analysis_task(task_id) ON DELETE CASCADE, FOREIGN KEY (segment_id) REFERENCES transcript_segment(segment_id) ON DELETE CASCADE, INDEX idx_type (detail_type) );

3. 数据入库实现方案

3.1 Python数据库操作封装

import mysql.connector from mysql.connector import Error import json from typing import Dict, List class VoiceAnalysisDB: def __init__(self, config: Dict): self.config = config self.connection = None def __enter__(self): self.connect() return self def __exit__(self, exc_type, exc_val, exc_tb): self.disconnect() def connect(self): try: self.connection = mysql.connector.connect(**self.config) print("数据库连接成功") except Error as e: print(f"连接失败: {e}") raise def disconnect(self): if self.connection and self.connection.is_connected(): self.connection.close() print("数据库连接已关闭") def save_analysis(self, task_data: Dict, segments: List[Dict], details: List[Dict]): cursor = None try: cursor = self.connection.cursor() self.connection.start_transaction() # 插入任务数据 cursor.execute(""" INSERT INTO analysis_task (task_id, audio_file, duration, model_version, status, completed_at) VALUES (%s, %s, %s, %s, 'success', NOW()) """, ( task_data['task_id'], task_data['audio_file'], task_data['duration'], task_data['model_version'] )) # 批量插入分段数据 segment_values = [ ( task_data['task_id'], seg['segment_no'], seg['start'], seg['end'], seg['text'], seg.get('sentiment'), seg.get('confidence') ) for seg in segments ] cursor.executemany(""" INSERT INTO transcript_segment (task_id, segment_no, start_time, end_time, text_content, sentiment, confidence) VALUES (%s, %s, %s, %s, %s, %s, %s) """, segment_values) # 批量插入明细数据 if details: cursor.executemany(""" INSERT INTO analysis_detail (task_id, segment_id, detail_type, detail_value, confidence) VALUES (%s, %s, %s, %s, %s) """, details) self.connection.commit() print(f"成功保存任务 {task_data['task_id']}") except Error as e: self.connection.rollback() print(f"数据保存失败: {e}") raise finally: if cursor: cursor.close()

3.2 数据转换与入库流程

def process_future_police_result(json_data: Dict) -> Dict: """转换FUTURE POLICE的JSON结果为数据库格式""" task_data = { 'task_id': json_data.get('task_id', generate_uuid()), 'audio_file': json_data['audio_meta']['file_name'], 'duration': json_data['audio_meta']['duration'], 'model_version': json_data.get('model_version', 'unknown') } segments = [] for idx, seg in enumerate(json_data['segments']): segments.append({ 'segment_no': idx, 'start': seg['start'], 'end': seg['end'], 'text': seg['text'], 'sentiment': seg.get('sentiment', {}).get('label'), 'confidence': seg.get('sentiment', {}).get('score') }) details = [] for seg_idx, seg in enumerate(json_data['segments']): # 处理关键词 for kw in seg.get('keywords', []): details.append(( task_data['task_id'], None, # segment_id将在入库时填充 'keyword', kw['text'], kw.get('confidence') )) # 处理实体 for ent in seg.get('entities', []): details.append(( task_data['task_id'], None, 'entity', f"{ent['type']}:{ent['text']}", ent.get('confidence') )) return { 'task_data': task_data, 'segments': segments, 'details': details } # 使用示例 db_config = { 'host': 'localhost', 'database': 'voice_analysis', 'user': 'voice_user', 'password': 'your_password' } with VoiceAnalysisDB(db_config) as db: fp_result = load_future_police_result('result.json') # 你的数据加载函数 processed = process_future_police_result(fp_result) db.save_analysis(**processed)

4. 实用SQL查询示例

4.1 基础统计分析

-- 查询情感分布 SELECT sentiment, COUNT(*) as count FROM transcript_segment WHERE sentiment IS NOT NULL GROUP BY sentiment ORDER BY count DESC; -- 查找包含特定关键词的片段 SELECT s.task_id, s.segment_no, s.text_content, d.detail_value as keyword FROM transcript_segment s JOIN analysis_detail d ON s.segment_id = d.segment_id WHERE d.detail_type = 'keyword' AND d.detail_value LIKE '%投诉%';

4.2 高级分析查询

-- 情感变化趋势分析（按时间窗口） SELECT FLOOR(start_time/60) as minute_window, AVG(CASE WHEN sentiment = 'positive' THEN 1 ELSE 0 END) as positive_rate, AVG(CASE WHEN sentiment = 'negative' THEN 1 ELSE 0 END) as negative_rate FROM transcript_segment WHERE task_id = 'your_task_id' GROUP BY FLOOR(start_time/60) ORDER BY minute_window; -- 关键词与情感关联分析 SELECT d.detail_value as keyword, COUNT(*) as total, AVG(CASE WHEN s.sentiment = 'positive' THEN 1 ELSE 0 END) as positive_ratio FROM analysis_detail d JOIN transcript_segment s ON d.segment_id = s.segment_id WHERE d.detail_type = 'keyword' GROUP BY d.detail_value HAVING total > 3 ORDER BY positive_ratio DESC;