FUTURE POLICE与数据库集成将语音解构结果存入MySQL并进行查询分析你是不是也遇到过这样的场景手头有一大堆会议录音、客服对话或者访谈音频用FUTURE POLICE处理完后生成了一大堆结构化的文本、情感标签和说话人信息。这些数据散落在各个JSON文件里想找个历史记录得翻半天更别提做跨文件的分析了。数据是宝藏但散落一地就成了垃圾。今天咱们就来聊聊怎么把这些零散的语音解构结果规规矩矩地存进MySQL数据库里让它变成真正能查、能看、能分析的数据资产。这就像给一堆珍贵的照片建一个带标签的电子相册以后想看哪张、按什么条件找都变得轻而易举。1. 为什么要把语音解构结果存进数据库你可能觉得处理完的JSON文件直接存硬盘不就行了刚开始数据量小的时候确实可以但一旦文件多了问题就来了。想象一下你想分析过去三个月所有客服通话中客户表达“不满意”情绪的高峰时段。如果数据都在文件里你得写个脚本遍历成百上千个JSON一个个打开、解析、过滤效率低不说还容易出错。但如果数据在MySQL里一句SQL查询就能搞定。把数据存进数据库核心是为了三件事持久化、结构化和可分析。持久化保证数据不丢结构化让杂乱的数据变得规整可分析则是赋予数据灵魂让你能从数据里挖出真金白银的业务洞察。接下来我们就一步步看看怎么实现。2. 设计你的语音数据仓库数据库表结构在动手写代码之前得先想好数据怎么放。一个好的表结构是高效查询和分析的基础。根据FUTURE POLICE典型的输出我们可以设计几个核心表。2.1 核心表设计思路我们的数据模型主要围绕一次音频处理任务展开。一次任务比如处理一个小时的会议录音会产生一条总的元数据记录以及多条具体的语音片段记录。我建议设计三张主表关系比较清晰audio_processing_jobs: 存放每次处理任务的元信息。audio_segments: 存放每个语音片段如一句话的详细解构结果。speakers: 存放说话人信息如果FUTURE POLICE支持说话人分离。2.2 表结构定义下面是用SQL语句定义的表结构你可以直接在MySQL中执行。-- 1. 音频处理任务表 CREATE TABLE audio_processing_jobs ( job_id INT AUTO_INCREMENT PRIMARY KEY, audio_file_name VARCHAR(255) NOT NULL COMMENT 原始音频文件名, file_path VARCHAR(500) COMMENT 文件存储路径, file_size BIGINT COMMENT 文件大小(字节), duration_seconds FLOAT COMMENT 音频时长(秒), model_version VARCHAR(50) COMMENT 使用的FUTURE POLICE模型版本, processing_status ENUM(pending, processing, completed, failed) DEFAULT pending, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, completed_at TIMESTAMP NULL, additional_metadata JSON COMMENT 其他原始元数据以JSON格式存储 ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT音频处理任务总表; -- 2. 说话人表如果分析结果包含说话人信息 CREATE TABLE speakers ( speaker_id INT AUTO_INCREMENT PRIMARY KEY, job_id INT NOT NULL, speaker_label VARCHAR(100) COMMENT 说话人标签如spk_0, spk_1, -- 未来可扩展 speaker_name, gender, age_group 等如果模型能识别 FOREIGN KEY (job_id) REFERENCES audio_processing_jobs(job_id) ON DELETE CASCADE ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT说话人信息表; -- 3. 音频片段表核心表 CREATE TABLE audio_segments ( segment_id INT AUTO_INCREMENT PRIMARY KEY, job_id INT NOT NULL, speaker_id INT NULL COMMENT 关联说话人可为空, start_time FLOAT NOT NULL COMMENT 片段开始时间(秒), end_time FLOAT NOT NULL COMMENT 片段结束时间(秒), text_content TEXT NOT NULL COMMENT 识别或解构出的文本, sentiment_label VARCHAR(50) COMMENT 情感标签如positive, negative, neutral, sentiment_score FLOAT COMMENT 情感置信度分数, keywords JSON COMMENT 提取的关键词JSON数组格式如[需求, 延期, 解决], entities JSON COMMENT 识别的实体JSON数组格式如[{type: PERSON, text: 张三}], created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (job_id) REFERENCES audio_processing_jobs(job_id) ON DELETE CASCADE, FOREIGN KEY (speaker_id) REFERENCES speakers(speaker_id) ON DELETE SET NULL, INDEX idx_job_id (job_id), INDEX idx_sentiment (sentiment_label), INDEX idx_time_range (start_time, end_time) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT音频片段详细结果表;设计要点说明关系清晰audio_segments通过job_id关联到总任务通过可选的speaker_id关联到说话人。JSON字段的妙用对于结构可能变化的keywords和entities我们使用了JSON类型。MySQL支持对JSON字段进行查询这样既保持了灵活性又不失查询能力。索引是关键在job_id,sentiment_label,time_range上建立了索引。当你的数据量达到几十万、上百万条时这些索引能让你的查询速度飞起来。注释不能少每个字段的COMMENT能帮你和你的队友快速理解字段含义是良好的开发习惯。3. 从代码到数据库数据持久化实战表建好了接下来就是把FUTURE POLICE处理后的数据灌进去。这里我用Python和SQLAlchemy这个ORM工具来演示因为它写起来更直观像操作Python对象一样操作数据库。3.1 环境准备与连接数据库首先确保安装了必要的库。pip install sqlalchemy pymysql然后我们创建一个数据库连接和映射表结构的Python类。# db_models.py from sqlalchemy import create_engine, Column, Integer, String, Float, Text, TIMESTAMP, JSON, Enum, ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import relationship, sessionmaker from datetime import datetime import json # 1. 创建数据库连接引擎 # 替换为你自己的数据库信息 DATABASE_URL mysqlpymysql://username:passwordlocalhost:3306/voice_analysis_db engine create_engine(DATABASE_URL, echoFalse) # echoTrue 可以查看执行的SQL调试用 Base declarative_base() # 2. 定义Python类对应数据库表 class AudioProcessingJob(Base): __tablename__ audio_processing_jobs job_id Column(Integer, primary_keyTrue, autoincrementTrue) audio_file_name Column(String(255), nullableFalse) file_path Column(String(500)) file_size Column(Integer) # 对应SQL的BIGINT duration_seconds Column(Float) model_version Column(String(50)) processing_status Column(Enum(pending, processing, completed, failed), defaultpending) created_at Column(TIMESTAMP, defaultdatetime.utcnow) completed_at Column(TIMESTAMP) additional_metadata Column(JSON) # 定义关系一对多 segments relationship(AudioSegment, back_populatesjob) speakers relationship(Speaker, back_populatesjob) class Speaker(Base): __tablename__ speakers speaker_id Column(Integer, primary_keyTrue, autoincrementTrue) job_id Column(Integer, ForeignKey(audio_processing_jobs.job_id, ondeleteCASCADE), nullableFalse) speaker_label Column(String(100)) # 定义关系 job relationship(AudioProcessingJob, back_populatesspeakers) segments relationship(AudioSegment, back_populatesspeaker) class AudioSegment(Base): __tablename__ audio_segments segment_id Column(Integer, primary_keyTrue, autoincrementTrue) job_id Column(Integer, ForeignKey(audio_processing_jobs.job_id, ondeleteCASCADE), nullableFalse) speaker_id Column(Integer, ForeignKey(speakers.speaker_id, ondeleteSET NULL)) start_time Column(Float, nullableFalse) end_time Column(Float, nullableFalse) text_content Column(Text, nullableFalse) sentiment_label Column(String(50)) sentiment_score Column(Float) keywords Column(JSON) entities Column(JSON) created_at Column(TIMESTAMP, defaultdatetime.utcnow) # 定义关系 job relationship(AudioProcessingJob, back_populatessegments) speaker relationship(Speaker, back_populatessegments) # 3. 创建所有表如果表不存在 Base.metadata.create_all(engine) # 4. 创建会话工厂 SessionLocal sessionmaker(bindengine)3.2 模拟数据插入一个完整的流程假设我们已经从FUTURE POLICE拿到了一份处理结果通常是JSON格式。下面我们模拟一次完整的数据插入过程。# data_insert_demo.py from db_models import SessionLocal, AudioProcessingJob, Speaker, AudioSegment import json def insert_processing_result(audio_file_path, future_police_result_json): 将一次FUTURE POLICE的处理结果存入数据库。 session SessionLocal() try: # 1. 解析FUTURE POLICE的JSON结果 result_data json.loads(future_police_result_json) # 假设已经是字典 # 实际中你需要根据FUTURE POLICE的实际输出格式来解析 # 这里我们模拟一个结构 metadata result_data.get(metadata, {}) segments_data result_data.get(segments, []) speakers_data result_data.get(speakers, []) # 假设有说话人信息 # 2. 创建并插入主任务记录 new_job AudioProcessingJob( audio_file_nameaudio_file_path.split(/)[-1], file_pathaudio_file_path, file_sizemetadata.get(file_size), duration_secondsmetadata.get(duration), model_versionmetadata.get(model_version, v1.0), processing_statuscompleted, completed_atdatetime.utcnow(), additional_metadatametadata # 整个元数据作为JSON存入 ) session.add(new_job) session.flush() # 获取新生成的job_id # 3. 插入说话人信息如果有 speaker_map {} # 用于映射说话人标签到数据库ID for spk in speakers_data: new_speaker Speaker( job_idnew_job.job_id, speaker_labelspk.get(label) ) session.add(new_speaker) session.flush() speaker_map[spk.get(label)] new_speaker.speaker_id # 4. 插入所有音频片段 for seg in segments_data: new_segment AudioSegment( job_idnew_job.job_id, speaker_idspeaker_map.get(seg.get(speaker)), # 关联说话人 start_timeseg.get(start), end_timeseg.get(end), text_contentseg.get(text, ), sentiment_labelseg.get(sentiment, {}).get(label), sentiment_scoreseg.get(sentiment, {}).get(score), keywordsjson.dumps(seg.get(keywords, [])), # 列表转JSON字符串 entitiesjson.dumps(seg.get(entities, [])) # 列表转JSON字符串 ) session.add(new_segment) # 5. 提交事务 session.commit() print(f成功插入任务 {new_job.job_id}, 包含 {len(segments_data)} 个片段。) return new_job.job_id except Exception as e: session.rollback() print(f数据插入失败: {e}) return None finally: session.close() # 模拟调用 if __name__ __main__: # 模拟一份FUTURE POLICE的输出 mock_result { metadata: { file_size: 1024000, duration: 3600.5, model_version: future_police_v2 }, speakers: [ {label: spk_0}, {label: spk_1} ], segments: [ { start: 10.5, end: 25.3, speaker: spk_0, text: 大家好我们开始今天的项目评审会议。, sentiment: {label: neutral, score: 0.05}, keywords: [项目, 评审, 会议], entities: [{type: EVENT, text: 项目评审会议}] }, { start: 26.1, end: 45.8, speaker: spk_1, text: 我觉得当前版本的核心功能已经非常完善了用户体验很好, sentiment: {label: positive, score: 0.92}, keywords: [版本, 核心功能, 用户体验], entities: [] } # ... 更多片段 ] } job_id insert_processing_result( audio_file_path/data/meetings/project_review_20231010.mp3, future_police_result_jsonjson.dumps(mock_result) )这段代码模拟了从解析结果到存入数据库的完整链路。在实际应用中你需要将mock_result替换为调用FUTURE POLICE API或解析其输出文件得到的真实数据。4. 让数据说话SQL查询与分析实战数据存进去只是第一步让数据产生价值才是目的。有了结构化的数据我们可以轻松地进行各种维度的分析。4.1 基础查询快速定位信息首先我们通过SQLAlchemy会话执行一些基础查询。# query_demo.py from db_models import SessionLocal, AudioProcessingJob, AudioSegment from sqlalchemy import func, desc session SessionLocal() # 示例1查询某个文件的所有对话片段 file_name project_review_20231010.mp3 job session.query(AudioProcessingJob).filter_by(audio_file_namefile_name).first() if job: segments session.query(AudioSegment).filter_by(job_idjob.job_id).order_by(AudioSegment.start_time).all() print(f文件 {file_name} 共有 {len(segments)} 个片段。) for seg in segments[:3]: # 打印前3个片段 print(f [{seg.start_time:.1f}s - {seg.end_time:.1f}s]: {seg.text_content[:50]}...) # 示例2查找所有包含负面情绪的片段 negative_segments (session.query(AudioSegment) .filter(AudioSegment.sentiment_label negative) .order_by(desc(AudioSegment.sentiment_score)) # 按负面程度排序 .limit(5).all()) print(f\n负面情绪最强烈的5个片段) for seg in negative_segments: print(f 分数{seg.sentiment_score:.2f}: {seg.text_content[:60]}...)4.2 进阶分析挖掘业务洞察更复杂的分析我们可以直接使用SQL的聚合和分组功能。以下是一些实用的分析场景。# advanced_analysis.py from db_models import SessionLocal, AudioSegment, AudioProcessingJob from sqlalchemy import func, extract, case, text import pandas as pd # 可选用于更复杂的数据分析 session SessionLocal() # 分析1情感趋势分析按任务/时间 # 计算每个处理任务中积极、中性、消极情感的占比 sentiment_analysis (session.query( AudioProcessingJob.audio_file_name, func.count(AudioSegment.segment_id).label(total_segments), func.sum(case((AudioSegment.sentiment_label positive, 1), else_0)).label(positive_count), func.sum(case((AudioSegment.sentiment_label negative, 1), else_0)).label(negative_count), func.sum(case((AudioSegment.sentiment_label neutral, 1), else_0)).label(neutral_count) ) .join(AudioSegment, AudioProcessingJob.job_id AudioSegment.job_id) .group_by(AudioProcessingJob.job_id, AudioProcessingJob.audio_file_name) .all()) print(各文件情感分布) for row in sentiment_analysis: pos_rate (row.positive_count / row.total_segments * 100) if row.total_segments else 0 neg_rate (row.negative_count / row.total_segments * 100) if row.total_segments else 0 print(f {row.audio_file_name}: 总片段{row.total_segments}, 积极{pos_rate:.1f}%, 消极{neg_rate:.1f}%) # 分析2高频词统计从JSON字段中提取 # 注意MySQL 5.7 支持JSON_EXTRACT函数这里演示SQL写法 # 由于SQLAlchemy对JSON数组的跨行展开支持有限复杂JSON分析有时直接用SQL更简单 high_freq_sql SELECT js.keyword, COUNT(*) as frequency FROM audio_segments s, JSON_TABLE(s.keywords, $[*] COLUMNS (keyword VARCHAR(50) PATH $)) AS js WHERE s.job_id :job_id GROUP BY js.keyword ORDER BY frequency DESC LIMIT 10; # 假设我们要分析 job_id1 的任务 result_proxy session.execute(text(high_freq_sql), {job_id: 1}) top_keywords result_proxy.fetchall() print(f\n任务1的高频关键词) for kw, freq in top_keywords: print(f {kw}: {freq}次) # 分析3说话人活跃度分析如果启用了说话人分离 speaker_activity_sql SELECT sp.speaker_label, COUNT(s.segment_id) as segment_count, SUM(s.end_time - s.start_time) as total_speaking_time, AVG(s.sentiment_score) as avg_sentiment FROM audio_segments s JOIN speakers sp ON s.speaker_id sp.speaker_id WHERE s.job_id :job_id GROUP BY sp.speaker_id, sp.speaker_label ORDER BY total_speaking_time DESC; result_proxy session.execute(text(speaker_activity_sql), {job_id: 1}) speaker_stats result_proxy.fetchall() print(f\n任务1的说话人活跃度) for speaker, count, duration, avg_sent in speaker_stats: print(f {speaker}: 发言{count}次共{duration:.1f}秒平均情感分{avg_sent:.2f}) session.close()这些分析示例只是冰山一角。一旦数据入库你可以结合业务需求轻松实现更多分析比如热点话题追踪分析不同时间段关键词的出现频率变化。客户情绪监控对客服通话进行实时情感打分和预警。会议效率评估分析会议中有效发言时长与沉默/重复段落的比例。5. 总结走完这一套流程你会发现原本杂乱无章的语音解构数据已经变成了一个规整、随时待命的数据资产库。从设计表结构时的深思熟虑到用SQLAlchemy优雅地插入数据再到用SQL进行多维度的挖掘分析每一步都是在为数据赋能。实际用下来最大的感受是“省心”和“强大”。省心在于再也不用担心文件管理混乱所有历史记录一目了然。强大在于分析维度可以随心所欲地扩展今天想看情感趋势明天想统计热词写个查询就能搞定不用再重写解析脚本。当然这只是个起点。当数据量进一步增长你可能需要考虑更高级的特性比如对海量文本内容进行全文索引可以用MySQL的全文索引或者引入Elasticsearch或者建立定期分析报表甚至做实时情感仪表盘。但无论如何把数据规规矩矩地存进关系型数据库都是构建这些上层应用最坚实、最正确的一步。如果你正准备管理越来越多的语音数据不妨就从设计这几张表开始试试。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
FUTURE POLICE与数据库集成：将语音解构结果存入MySQL并进行查询分析
FUTURE POLICE与数据库集成将语音解构结果存入MySQL并进行查询分析你是不是也遇到过这样的场景手头有一大堆会议录音、客服对话或者访谈音频用FUTURE POLICE处理完后生成了一大堆结构化的文本、情感标签和说话人信息。这些数据散落在各个JSON文件里想找个历史记录得翻半天更别提做跨文件的分析了。数据是宝藏但散落一地就成了垃圾。今天咱们就来聊聊怎么把这些零散的语音解构结果规规矩矩地存进MySQL数据库里让它变成真正能查、能看、能分析的数据资产。这就像给一堆珍贵的照片建一个带标签的电子相册以后想看哪张、按什么条件找都变得轻而易举。1. 为什么要把语音解构结果存进数据库你可能觉得处理完的JSON文件直接存硬盘不就行了刚开始数据量小的时候确实可以但一旦文件多了问题就来了。想象一下你想分析过去三个月所有客服通话中客户表达“不满意”情绪的高峰时段。如果数据都在文件里你得写个脚本遍历成百上千个JSON一个个打开、解析、过滤效率低不说还容易出错。但如果数据在MySQL里一句SQL查询就能搞定。把数据存进数据库核心是为了三件事持久化、结构化和可分析。持久化保证数据不丢结构化让杂乱的数据变得规整可分析则是赋予数据灵魂让你能从数据里挖出真金白银的业务洞察。接下来我们就一步步看看怎么实现。2. 设计你的语音数据仓库数据库表结构在动手写代码之前得先想好数据怎么放。一个好的表结构是高效查询和分析的基础。根据FUTURE POLICE典型的输出我们可以设计几个核心表。2.1 核心表设计思路我们的数据模型主要围绕一次音频处理任务展开。一次任务比如处理一个小时的会议录音会产生一条总的元数据记录以及多条具体的语音片段记录。我建议设计三张主表关系比较清晰audio_processing_jobs: 存放每次处理任务的元信息。audio_segments: 存放每个语音片段如一句话的详细解构结果。speakers: 存放说话人信息如果FUTURE POLICE支持说话人分离。2.2 表结构定义下面是用SQL语句定义的表结构你可以直接在MySQL中执行。-- 1. 音频处理任务表 CREATE TABLE audio_processing_jobs ( job_id INT AUTO_INCREMENT PRIMARY KEY, audio_file_name VARCHAR(255) NOT NULL COMMENT 原始音频文件名, file_path VARCHAR(500) COMMENT 文件存储路径, file_size BIGINT COMMENT 文件大小(字节), duration_seconds FLOAT COMMENT 音频时长(秒), model_version VARCHAR(50) COMMENT 使用的FUTURE POLICE模型版本, processing_status ENUM(pending, processing, completed, failed) DEFAULT pending, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, completed_at TIMESTAMP NULL, additional_metadata JSON COMMENT 其他原始元数据以JSON格式存储 ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT音频处理任务总表; -- 2. 说话人表如果分析结果包含说话人信息 CREATE TABLE speakers ( speaker_id INT AUTO_INCREMENT PRIMARY KEY, job_id INT NOT NULL, speaker_label VARCHAR(100) COMMENT 说话人标签如spk_0, spk_1, -- 未来可扩展 speaker_name, gender, age_group 等如果模型能识别 FOREIGN KEY (job_id) REFERENCES audio_processing_jobs(job_id) ON DELETE CASCADE ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT说话人信息表; -- 3. 音频片段表核心表 CREATE TABLE audio_segments ( segment_id INT AUTO_INCREMENT PRIMARY KEY, job_id INT NOT NULL, speaker_id INT NULL COMMENT 关联说话人可为空, start_time FLOAT NOT NULL COMMENT 片段开始时间(秒), end_time FLOAT NOT NULL COMMENT 片段结束时间(秒), text_content TEXT NOT NULL COMMENT 识别或解构出的文本, sentiment_label VARCHAR(50) COMMENT 情感标签如positive, negative, neutral, sentiment_score FLOAT COMMENT 情感置信度分数, keywords JSON COMMENT 提取的关键词JSON数组格式如[需求, 延期, 解决], entities JSON COMMENT 识别的实体JSON数组格式如[{type: PERSON, text: 张三}], created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (job_id) REFERENCES audio_processing_jobs(job_id) ON DELETE CASCADE, FOREIGN KEY (speaker_id) REFERENCES speakers(speaker_id) ON DELETE SET NULL, INDEX idx_job_id (job_id), INDEX idx_sentiment (sentiment_label), INDEX idx_time_range (start_time, end_time) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT音频片段详细结果表;设计要点说明关系清晰audio_segments通过job_id关联到总任务通过可选的speaker_id关联到说话人。JSON字段的妙用对于结构可能变化的keywords和entities我们使用了JSON类型。MySQL支持对JSON字段进行查询这样既保持了灵活性又不失查询能力。索引是关键在job_id,sentiment_label,time_range上建立了索引。当你的数据量达到几十万、上百万条时这些索引能让你的查询速度飞起来。注释不能少每个字段的COMMENT能帮你和你的队友快速理解字段含义是良好的开发习惯。3. 从代码到数据库数据持久化实战表建好了接下来就是把FUTURE POLICE处理后的数据灌进去。这里我用Python和SQLAlchemy这个ORM工具来演示因为它写起来更直观像操作Python对象一样操作数据库。3.1 环境准备与连接数据库首先确保安装了必要的库。pip install sqlalchemy pymysql然后我们创建一个数据库连接和映射表结构的Python类。# db_models.py from sqlalchemy import create_engine, Column, Integer, String, Float, Text, TIMESTAMP, JSON, Enum, ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import relationship, sessionmaker from datetime import datetime import json # 1. 创建数据库连接引擎 # 替换为你自己的数据库信息 DATABASE_URL mysqlpymysql://username:passwordlocalhost:3306/voice_analysis_db engine create_engine(DATABASE_URL, echoFalse) # echoTrue 可以查看执行的SQL调试用 Base declarative_base() # 2. 定义Python类对应数据库表 class AudioProcessingJob(Base): __tablename__ audio_processing_jobs job_id Column(Integer, primary_keyTrue, autoincrementTrue) audio_file_name Column(String(255), nullableFalse) file_path Column(String(500)) file_size Column(Integer) # 对应SQL的BIGINT duration_seconds Column(Float) model_version Column(String(50)) processing_status Column(Enum(pending, processing, completed, failed), defaultpending) created_at Column(TIMESTAMP, defaultdatetime.utcnow) completed_at Column(TIMESTAMP) additional_metadata Column(JSON) # 定义关系一对多 segments relationship(AudioSegment, back_populatesjob) speakers relationship(Speaker, back_populatesjob) class Speaker(Base): __tablename__ speakers speaker_id Column(Integer, primary_keyTrue, autoincrementTrue) job_id Column(Integer, ForeignKey(audio_processing_jobs.job_id, ondeleteCASCADE), nullableFalse) speaker_label Column(String(100)) # 定义关系 job relationship(AudioProcessingJob, back_populatesspeakers) segments relationship(AudioSegment, back_populatesspeaker) class AudioSegment(Base): __tablename__ audio_segments segment_id Column(Integer, primary_keyTrue, autoincrementTrue) job_id Column(Integer, ForeignKey(audio_processing_jobs.job_id, ondeleteCASCADE), nullableFalse) speaker_id Column(Integer, ForeignKey(speakers.speaker_id, ondeleteSET NULL)) start_time Column(Float, nullableFalse) end_time Column(Float, nullableFalse) text_content Column(Text, nullableFalse) sentiment_label Column(String(50)) sentiment_score Column(Float) keywords Column(JSON) entities Column(JSON) created_at Column(TIMESTAMP, defaultdatetime.utcnow) # 定义关系 job relationship(AudioProcessingJob, back_populatessegments) speaker relationship(Speaker, back_populatessegments) # 3. 创建所有表如果表不存在 Base.metadata.create_all(engine) # 4. 创建会话工厂 SessionLocal sessionmaker(bindengine)3.2 模拟数据插入一个完整的流程假设我们已经从FUTURE POLICE拿到了一份处理结果通常是JSON格式。下面我们模拟一次完整的数据插入过程。# data_insert_demo.py from db_models import SessionLocal, AudioProcessingJob, Speaker, AudioSegment import json def insert_processing_result(audio_file_path, future_police_result_json): 将一次FUTURE POLICE的处理结果存入数据库。 session SessionLocal() try: # 1. 解析FUTURE POLICE的JSON结果 result_data json.loads(future_police_result_json) # 假设已经是字典 # 实际中你需要根据FUTURE POLICE的实际输出格式来解析 # 这里我们模拟一个结构 metadata result_data.get(metadata, {}) segments_data result_data.get(segments, []) speakers_data result_data.get(speakers, []) # 假设有说话人信息 # 2. 创建并插入主任务记录 new_job AudioProcessingJob( audio_file_nameaudio_file_path.split(/)[-1], file_pathaudio_file_path, file_sizemetadata.get(file_size), duration_secondsmetadata.get(duration), model_versionmetadata.get(model_version, v1.0), processing_statuscompleted, completed_atdatetime.utcnow(), additional_metadatametadata # 整个元数据作为JSON存入 ) session.add(new_job) session.flush() # 获取新生成的job_id # 3. 插入说话人信息如果有 speaker_map {} # 用于映射说话人标签到数据库ID for spk in speakers_data: new_speaker Speaker( job_idnew_job.job_id, speaker_labelspk.get(label) ) session.add(new_speaker) session.flush() speaker_map[spk.get(label)] new_speaker.speaker_id # 4. 插入所有音频片段 for seg in segments_data: new_segment AudioSegment( job_idnew_job.job_id, speaker_idspeaker_map.get(seg.get(speaker)), # 关联说话人 start_timeseg.get(start), end_timeseg.get(end), text_contentseg.get(text, ), sentiment_labelseg.get(sentiment, {}).get(label), sentiment_scoreseg.get(sentiment, {}).get(score), keywordsjson.dumps(seg.get(keywords, [])), # 列表转JSON字符串 entitiesjson.dumps(seg.get(entities, [])) # 列表转JSON字符串 ) session.add(new_segment) # 5. 提交事务 session.commit() print(f成功插入任务 {new_job.job_id}, 包含 {len(segments_data)} 个片段。) return new_job.job_id except Exception as e: session.rollback() print(f数据插入失败: {e}) return None finally: session.close() # 模拟调用 if __name__ __main__: # 模拟一份FUTURE POLICE的输出 mock_result { metadata: { file_size: 1024000, duration: 3600.5, model_version: future_police_v2 }, speakers: [ {label: spk_0}, {label: spk_1} ], segments: [ { start: 10.5, end: 25.3, speaker: spk_0, text: 大家好我们开始今天的项目评审会议。, sentiment: {label: neutral, score: 0.05}, keywords: [项目, 评审, 会议], entities: [{type: EVENT, text: 项目评审会议}] }, { start: 26.1, end: 45.8, speaker: spk_1, text: 我觉得当前版本的核心功能已经非常完善了用户体验很好, sentiment: {label: positive, score: 0.92}, keywords: [版本, 核心功能, 用户体验], entities: [] } # ... 更多片段 ] } job_id insert_processing_result( audio_file_path/data/meetings/project_review_20231010.mp3, future_police_result_jsonjson.dumps(mock_result) )这段代码模拟了从解析结果到存入数据库的完整链路。在实际应用中你需要将mock_result替换为调用FUTURE POLICE API或解析其输出文件得到的真实数据。4. 让数据说话SQL查询与分析实战数据存进去只是第一步让数据产生价值才是目的。有了结构化的数据我们可以轻松地进行各种维度的分析。4.1 基础查询快速定位信息首先我们通过SQLAlchemy会话执行一些基础查询。# query_demo.py from db_models import SessionLocal, AudioProcessingJob, AudioSegment from sqlalchemy import func, desc session SessionLocal() # 示例1查询某个文件的所有对话片段 file_name project_review_20231010.mp3 job session.query(AudioProcessingJob).filter_by(audio_file_namefile_name).first() if job: segments session.query(AudioSegment).filter_by(job_idjob.job_id).order_by(AudioSegment.start_time).all() print(f文件 {file_name} 共有 {len(segments)} 个片段。) for seg in segments[:3]: # 打印前3个片段 print(f [{seg.start_time:.1f}s - {seg.end_time:.1f}s]: {seg.text_content[:50]}...) # 示例2查找所有包含负面情绪的片段 negative_segments (session.query(AudioSegment) .filter(AudioSegment.sentiment_label negative) .order_by(desc(AudioSegment.sentiment_score)) # 按负面程度排序 .limit(5).all()) print(f\n负面情绪最强烈的5个片段) for seg in negative_segments: print(f 分数{seg.sentiment_score:.2f}: {seg.text_content[:60]}...)4.2 进阶分析挖掘业务洞察更复杂的分析我们可以直接使用SQL的聚合和分组功能。以下是一些实用的分析场景。# advanced_analysis.py from db_models import SessionLocal, AudioSegment, AudioProcessingJob from sqlalchemy import func, extract, case, text import pandas as pd # 可选用于更复杂的数据分析 session SessionLocal() # 分析1情感趋势分析按任务/时间 # 计算每个处理任务中积极、中性、消极情感的占比 sentiment_analysis (session.query( AudioProcessingJob.audio_file_name, func.count(AudioSegment.segment_id).label(total_segments), func.sum(case((AudioSegment.sentiment_label positive, 1), else_0)).label(positive_count), func.sum(case((AudioSegment.sentiment_label negative, 1), else_0)).label(negative_count), func.sum(case((AudioSegment.sentiment_label neutral, 1), else_0)).label(neutral_count) ) .join(AudioSegment, AudioProcessingJob.job_id AudioSegment.job_id) .group_by(AudioProcessingJob.job_id, AudioProcessingJob.audio_file_name) .all()) print(各文件情感分布) for row in sentiment_analysis: pos_rate (row.positive_count / row.total_segments * 100) if row.total_segments else 0 neg_rate (row.negative_count / row.total_segments * 100) if row.total_segments else 0 print(f {row.audio_file_name}: 总片段{row.total_segments}, 积极{pos_rate:.1f}%, 消极{neg_rate:.1f}%) # 分析2高频词统计从JSON字段中提取 # 注意MySQL 5.7 支持JSON_EXTRACT函数这里演示SQL写法 # 由于SQLAlchemy对JSON数组的跨行展开支持有限复杂JSON分析有时直接用SQL更简单 high_freq_sql SELECT js.keyword, COUNT(*) as frequency FROM audio_segments s, JSON_TABLE(s.keywords, $[*] COLUMNS (keyword VARCHAR(50) PATH $)) AS js WHERE s.job_id :job_id GROUP BY js.keyword ORDER BY frequency DESC LIMIT 10; # 假设我们要分析 job_id1 的任务 result_proxy session.execute(text(high_freq_sql), {job_id: 1}) top_keywords result_proxy.fetchall() print(f\n任务1的高频关键词) for kw, freq in top_keywords: print(f {kw}: {freq}次) # 分析3说话人活跃度分析如果启用了说话人分离 speaker_activity_sql SELECT sp.speaker_label, COUNT(s.segment_id) as segment_count, SUM(s.end_time - s.start_time) as total_speaking_time, AVG(s.sentiment_score) as avg_sentiment FROM audio_segments s JOIN speakers sp ON s.speaker_id sp.speaker_id WHERE s.job_id :job_id GROUP BY sp.speaker_id, sp.speaker_label ORDER BY total_speaking_time DESC; result_proxy session.execute(text(speaker_activity_sql), {job_id: 1}) speaker_stats result_proxy.fetchall() print(f\n任务1的说话人活跃度) for speaker, count, duration, avg_sent in speaker_stats: print(f {speaker}: 发言{count}次共{duration:.1f}秒平均情感分{avg_sent:.2f}) session.close()这些分析示例只是冰山一角。一旦数据入库你可以结合业务需求轻松实现更多分析比如热点话题追踪分析不同时间段关键词的出现频率变化。客户情绪监控对客服通话进行实时情感打分和预警。会议效率评估分析会议中有效发言时长与沉默/重复段落的比例。5. 总结走完这一套流程你会发现原本杂乱无章的语音解构数据已经变成了一个规整、随时待命的数据资产库。从设计表结构时的深思熟虑到用SQLAlchemy优雅地插入数据再到用SQL进行多维度的挖掘分析每一步都是在为数据赋能。实际用下来最大的感受是“省心”和“强大”。省心在于再也不用担心文件管理混乱所有历史记录一目了然。强大在于分析维度可以随心所欲地扩展今天想看情感趋势明天想统计热词写个查询就能搞定不用再重写解析脚本。当然这只是个起点。当数据量进一步增长你可能需要考虑更高级的特性比如对海量文本内容进行全文索引可以用MySQL的全文索引或者引入Elasticsearch或者建立定期分析报表甚至做实时情感仪表盘。但无论如何把数据规规矩矩地存进关系型数据库都是构建这些上层应用最坚实、最正确的一步。如果你正准备管理越来越多的语音数据不妨就从设计这几张表开始试试。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
