当前位置：首页 > news >正文

基于CCMusic的音乐推荐系统开发：MySQL数据库集成实践

news 2026/5/11 22:27:23

基于CCMusic的音乐推荐系统开发：MySQL数据库集成实践

引言

音乐推荐系统已经成为现代音乐平台的核心功能，而如何高效存储和管理音乐数据是实现智能推荐的关键。今天我们将探讨如何将CCMusic音乐分类结果与MySQL数据库深度集成，构建一个实用且高效的音乐推荐系统。

想象一下这样的场景：你的音乐平台每天新增上千首歌曲，每首歌曲都经过CCMusic模型自动分类为不同流派。如何存储这些海量数据？如何快速查询用户偏好？如何实现个性化推荐？这些问题都可以通过合理的数据库设计和优化来解决。

1. 系统架构设计

1.1 整体架构概述

我们的音乐推荐系统采用分层架构设计，从音乐数据处理到最终的用户推荐，包含以下几个核心模块：

音乐数据处理层：使用CCMusic模型对音频文件进行特征提取和流派分类
数据存储层：MySQL数据库负责存储音乐元数据、分类结果和用户行为数据
推荐算法层：基于用户历史行为和音乐特征计算个性化推荐
应用服务层：提供RESTful API接口给前端应用调用

1.2 数据库设计原则

在设计数据库时，我们遵循以下几个关键原则：

规范化设计：减少数据冗余，确保数据一致性
读写分离：针对频繁的查询操作进行优化
索引策略：为常用查询字段建立合适的索引
分区设计：对大数据量表进行分区管理

2. 数据库表结构设计

2.1 核心表设计

-- 音乐信息表 CREATE TABLE music ( id INT AUTO_INCREMENT PRIMARY KEY, title VARCHAR(255) NOT NULL, artist VARCHAR(255) NOT NULL, duration INT NOT NULL, file_path VARCHAR(500) NOT NULL, upload_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, bpm INT, key_signature VARCHAR(10), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_artist (artist), INDEX idx_upload_time (upload_time) ) ENGINE=InnoDB; -- 音乐流派分类表 CREATE TABLE music_genre ( id INT AUTO_INCREMENT PRIMARY KEY, music_id INT NOT NULL, primary_genre VARCHAR(50) NOT NULL, secondary_genre VARCHAR(50), confidence_score FLOAT NOT NULL, classified_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (music_id) REFERENCES music(id) ON DELETE CASCADE, INDEX idx_primary_genre (primary_genre), INDEX idx_confidence (confidence_score) ) ENGINE=InnoDB; -- 用户信息表 CREATE TABLE users ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(100) UNIQUE NOT NULL, email VARCHAR(255) UNIQUE NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_login TIMESTAMP, preferences JSON, INDEX idx_username (username) ) ENGINE=InnoDB; -- 用户行为表 CREATE TABLE user_behavior ( id BIGINT AUTO_INCREMENT PRIMARY KEY, user_id INT NOT NULL, music_id INT NOT NULL, behavior_type ENUM('play', 'like', 'share', 'skip', 'complete') NOT NULL, behavior_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, duration_played INT, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE, FOREIGN KEY (music_id) REFERENCES music(id) ON DELETE CASCADE, INDEX idx_user_behavior (user_id, behavior_type, behavior_time), INDEX idx_music_behavior (music_id, behavior_type) ) ENGINE=InnoDB; -- 推荐结果表 CREATE TABLE recommendations ( id BIGINT AUTO_INCREMENT PRIMARY KEY, user_id INT NOT NULL, music_id INT NOT NULL, recommendation_score FLOAT NOT NULL, recommended_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, reason VARCHAR(255), FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE, FOREIGN KEY (music_id) REFERENCES music(id) ON DELETE CASCADE, INDEX idx_user_recommendation (user_id, recommended_at), INDEX idx_recommendation_score (recommendation_score) ) ENGINE=InnoDB;

2.2 表关系说明

这些表之间通过外键关联，形成了一个完整的数据模型：

music表存储基本的音乐信息
music_genre表存储CCMusic的分类结果，与music表是一对多关系
users表存储用户基本信息
user_behavior表记录用户的所有交互行为
recommendations表存储生成的推荐结果

3. 数据入库实践

3.1 CCMusic分类结果入库

当CCMusic完成音乐分类后，我们需要将结果存储到数据库。以下是一个完整的入库示例：

import mysql.connector from mysql.connector import Error import json from datetime import datetime class MusicDatabase: def __init__(self, host, database, user, password): self.connection = mysql.connector.connect( host=host, database=database, user=user, password=password ) def insert_music_with_genre(self, music_data, genre_data): """插入音乐数据及分类结果""" try: cursor = self.connection.cursor() # 插入音乐基本信息 music_query = """ INSERT INTO music (title, artist, duration, file_path, bpm, key_signature) VALUES (%s, %s, %s, %s, %s, %s) """ music_values = ( music_data['title'], music_data['artist'], music_data['duration'], music_data['file_path'], music_data.get('bpm'), music_data.get('key_signature') ) cursor.execute(music_query, music_values) music_id = cursor.lastrowid # 插入分类结果 genre_query = """ INSERT INTO music_genre (music_id, primary_genre, secondary_genre, confidence_score) VALUES (%s, %s, %s, %s) """ genre_values = ( music_id, genre_data['primary_genre'], genre_data.get('secondary_genre'), genre_data['confidence_score'] ) cursor.execute(genre_query, genre_values) self.connection.commit() return music_id except Error as e: print(f"数据库插入错误: {e}") self.connection.rollback() return None finally: cursor.close() def batch_insert_music(self, music_list): """批量插入音乐数据""" try: cursor = self.connection.cursor() music_query = """ INSERT INTO music (title, artist, duration, file_path, bpm, key_signature) VALUES (%s, %s, %s, %s, %s, %s) """ genre_query = """ INSERT INTO music_genre (music_id, primary_genre, secondary_genre, confidence_score) VALUES (%s, %s, %s, %s) """ for music_data, genre_data in music_list: # 插入音乐信息 music_values = ( music_data['title'], music_data['artist'], music_data['duration'], music_data['file_path'], music_data.get('bpm'), music_data.get('key_signature') ) cursor.execute(music_query, music_values) music_id = cursor.lastrowid # 插入分类信息 genre_values = ( music_id, genre_data['primary_genre'], genre_data.get('secondary_genre'), genre_data['confidence_score'] ) cursor.execute(genre_query, genre_values) self.connection.commit() print(f"成功插入 {len(music_list)} 条记录") except Error as e: print(f"批量插入错误: {e}") self.connection.rollback() finally: cursor.close() # 使用示例 if __name__ == "__main__": db = MusicDatabase('localhost', 'music_db', 'username', 'password') # 单条数据插入 music_data = { 'title': '示例歌曲', 'artist': '示例歌手', 'duration': 240, 'file_path': '/music/sample.mp3', 'bpm': 120, 'key_signature': 'C major' } genre_data = { 'primary_genre': 'pop', 'secondary_genre': 'dance', 'confidence_score': 0.92 } music_id = db.insert_music_with_genre(music_data, genre_data) print(f"插入成功，音乐ID: {music_id}")

3.2 用户行为记录

用户行为数据是推荐系统的重要输入，需要高效记录：

class UserBehaviorLogger: def __init__(self, db_connection): self.connection = db_connection def log_behavior(self, user_id, music_id, behavior_type, duration_played=None): """记录用户行为""" try: cursor = self.connection.cursor() query = """ INSERT INTO user_behavior (user_id, music_id, behavior_type, duration_played) VALUES (%s, %s, %s, %s) """ values = (user_id, music_id, behavior_type, duration_played) cursor.execute(query, values) self.connection.commit() except Error as e: print(f"行为记录错误: {e}") self.connection.rollback() finally: cursor.close() def batch_log_behavior(self, behavior_list): """批量记录用户行为""" try: cursor = self.connection.cursor() query = """ INSERT INTO user_behavior (user_id, music_id, behavior_type, duration_played) VALUES (%s, %s, %s, %s) """ cursor.executemany(query, behavior_list) self.connection.commit() print(f"成功记录 {len(behavior_list)} 条行为数据") except Error as e: print(f"批量行为记录错误: {e}") self.connection.rollback() finally: cursor.close()

4. 查询优化与索引策略

4.1 常用查询优化

基于用户偏好的音乐推荐涉及大量复杂查询，以下是一些优化策略：

-- 创建复合索引优化常用查询 CREATE INDEX idx_user_genre_behavior ON user_behavior(user_id, behavior_type); CREATE INDEX idx_music_genre ON music_genre(music_id, primary_genre); CREATE INDEX idx_behavior_time ON user_behavior(behavior_time DESC); -- 优化用户偏好查询 SELECT m.*, mg.primary_genre, mg.confidence_score FROM music m JOIN music_genre mg ON m.id = mg.music_id JOIN user_behavior ub ON m.id = ub.music_id WHERE ub.user_id = 123 AND ub.behavior_type = 'like' AND ub.behavior_time > DATE_SUB(NOW(), INTERVAL 30 DAY) AND mg.primary_genre IN ('pop', 'rock') ORDER BY ub.behavior_time DESC LIMIT 50;

4.2 分区策略

对于大规模数据，采用分区策略可以显著提升查询性能：

-- 对用户行为表按时间进行分区 ALTER TABLE user_behavior PARTITION BY RANGE (YEAR(behavior_time)) ( PARTITION p2023 VALUES LESS THAN (2024), PARTITION p2024 VALUES LESS THAN (2025), PARTITION p2025 VALUES LESS THAN (2026), PARTITION p_future VALUES LESS THAN MAXVALUE ); -- 对音乐表按上传时间进行分区 ALTER TABLE music PARTITION BY RANGE (YEAR(upload_time)) ( PARTITION p2023 VALUES LESS THAN (2024), PARTITION p2024 VALUES LESS THAN (2025), PARTITION p2025 VALUES LESS THAN (2026), PARTITION p_future VALUES LESS THAN MAXVALUE );

5. 推荐算法实现

5.1 基于内容的推荐

利用CCMusic的分类结果实现基于内容的推荐：

class ContentBasedRecommender: def __init__(self, db_connection): self.connection = db_connection def get_user_genre_preference(self, user_id, days=30): """获取用户流派偏好""" try: cursor = self.connection.cursor(dictionary=True) query = """ SELECT mg.primary_genre, COUNT(*) as play_count, AVG(ub.duration_played) as avg_duration, MAX(ub.behavior_time) as last_played FROM user_behavior ub JOIN music_genre mg ON ub.music_id = mg.music_id WHERE ub.user_id = %s AND ub.behavior_type = 'play' AND ub.behavior_time > DATE_SUB(NOW(), INTERVAL %s DAY) GROUP BY mg.primary_genre ORDER BY play_count DESC, avg_duration DESC """ cursor.execute(query, (user_id, days)) results = cursor.fetchall() return results except Error as e: print(f"查询用户偏好错误: {e}") return [] finally: cursor.close() def recommend_by_genre(self, user_id, limit=10): """基于流派偏好推荐音乐""" try: # 获取用户偏好 preferences = self.get_user_genre_preference(user_id) if not preferences: return self.get_popular_music(limit) # 提取偏好流派 preferred_genres = [pref['primary_genre'] for pref in preferences[:3]] cursor = self.connection.cursor(dictionary=True) query = """ SELECT m.*, mg.primary_genre, mg.confidence_score FROM music m JOIN music_genre mg ON m.id = mg.music_id WHERE mg.primary_genre IN (%s) AND m.id NOT IN ( SELECT music_id FROM user_behavior WHERE user_id = %s AND behavior_type = 'play' ) ORDER BY mg.confidence_score DESC, m.upload_time DESC LIMIT %s """ # 构建IN查询参数 format_strings = ','.join(['%s'] * len(preferred_genres)) query = query % format_strings cursor.execute(query, preferred_genres + [user_id, limit]) results = cursor.fetchall() return results except Error as e: print(f"推荐查询错误: {e}") return [] finally: cursor.close()

5.2 协同过滤推荐

结合用户行为数据实现协同过滤：

class CollaborativeFilteringRecommender: def __init__(self, db_connection): self.connection = db_connection def find_similar_users(self, user_id, limit=5): """查找相似用户""" try: cursor = self.connection.cursor(dictionary=True) query = """ SELECT ub2.user_id, COUNT(*) as common_tracks, SUM(ub1.duration_played * ub2.duration_played) as similarity_score FROM user_behavior ub1 JOIN user_behavior ub2 ON ub1.music_id = ub2.music_id AND ub1.behavior_type = ub2.behavior_type AND ub2.user_id != ub1.user_id WHERE ub1.user_id = %s AND ub1.behavior_time > DATE_SUB(NOW(), INTERVAL 30 DAY) GROUP BY ub2.user_id ORDER BY similarity_score DESC LIMIT %s """ cursor.execute(query, (user_id, limit)) results = cursor.fetchall() return results except Error as e: print(f"查找相似用户错误: {e}") return [] finally: cursor.close() def recommend_from_similar_users(self, user_id, limit=10): """基于相似用户推荐""" try: similar_users = self.find_similar_users(user_id) if not similar_users: return [] similar_user_ids = [user['user_id'] for user in similar_users] cursor = self.connection.cursor(dictionary=True) query = """ SELECT m.*, mg.primary_genre, COUNT(*) as play_count FROM music m JOIN music_genre mg ON m.id = mg.music_id JOIN user_behavior ub ON m.id = ub.music_id WHERE ub.user_id IN (%s) AND ub.behavior_type = 'play' AND m.id NOT IN ( SELECT music_id FROM user_behavior WHERE user_id = %s ) GROUP BY m.id ORDER BY play_count DESC, mg.confidence_score DESC LIMIT %s """ format_strings = ','.join(['%s'] * len(similar_user_ids)) query = query % format_strings cursor.execute(query, similar_user_ids + [user_id, limit]) results = cursor.fetchall() return results except Error as e: print(f"协同推荐错误: {e}") return [] finally: cursor.close()

6. 性能监控与优化

6.1 查询性能监控

定期监控数据库性能，确保推荐系统响应迅速：

-- 查看慢查询日志 SHOW VARIABLES LIKE 'slow_query_log'; SHOW VARIABLES LIKE 'long_query_time'; -- 分析查询执行计划 EXPLAIN ANALYZE SELECT m.title, m.artist, mg.primary_genre FROM music m JOIN music_genre mg ON m.id = mg.music_id WHERE mg.primary_genre = 'pop' ORDER BY m.upload_time DESC LIMIT 100; -- 监控索引使用情况 SELECT OBJECT_NAME, INDEX_NAME, COUNT_READ, COUNT_FETCH FROM performance_schema.table_io_waits_summary_by_index_usage WHERE OBJECT_SCHEMA = 'music_db';

6.2 数据库维护策略

定期进行数据库维护，确保系统稳定运行：

class DatabaseMaintenance: def __init__(self, db_connection): self.connection = db_connection def optimize_tables(self): """优化数据库表""" try: cursor = self.connection.cursor() # 获取所有表名 cursor.execute("SHOW TABLES") tables = [table[0] for table in cursor.fetchall()] for table in tables: print(f"优化表: {table}") cursor.execute(f"OPTIMIZE TABLE {table}") print("所有表优化完成") except Error as e: print(f"表优化错误: {e}") finally: cursor.close() def cleanup_old_data(self, days=365): """清理旧数据""" try: cursor = self.connection.cursor() # 清理旧用户行为数据 delete_query = """ DELETE FROM user_behavior WHERE behavior_time < DATE_SUB(NOW(), INTERVAL %s DAY) """ cursor.execute(delete_query, (days,)) deleted_count = cursor.rowcount self.connection.commit() print(f"清理了 {deleted_count} 条旧行为记录") except Error as e: print(f"数据清理错误: {e}") self.connection.rollback() finally: cursor.close()

7. 实际应用案例

7.1 个性化推荐接口

实现一个完整的推荐API接口：

from flask import Flask, jsonify, request import mysql.connector from mysql.connector import Error app = Flask(__name__) def get_db_connection(): """获取数据库连接""" return mysql.connector.connect( host='localhost', database='music_db', user='username', password='password' ) @app.route('/recommendations/<int:user_id>', methods=['GET']) def get_recommendations(user_id): """获取个性化推荐""" try: connection = get_db_connection() # 获取基于内容的推荐 content_recommender = ContentBasedRecommender(connection) content_recs = content_recommender.recommend_by_genre(user_id, limit=5) # 获取协同过滤推荐 collab_recommender = CollaborativeFilteringRecommender(connection) collab_recs = collab_recommender.recommend_from_similar_users(user_id, limit=5) # 合并推荐结果 all_recommendations = content_recs + collab_recs # 去重并排序 seen_ids = set() unique_recommendations = [] for rec in all_recommendations: if rec['id'] not in seen_ids: seen_ids.add(rec['id']) unique_recommendations.append(rec) # 存储推荐结果 save_recommendations(connection, user_id, unique_recommendations) return jsonify({ 'success': True, 'recommendations': unique_recommendations[:10], 'count': len(unique_recommendations) }) except Error as e: return jsonify({'success': False, 'error': str(e)}) finally: if connection.is_connected(): connection.close() def save_recommendations(connection, user_id, recommendations): """保存推荐结果到数据库""" try: cursor = connection.cursor() # 先清除旧的推荐结果 delete_query = "DELETE FROM recommendations WHERE user_id = %s" cursor.execute(delete_query, (user_id,)) # 插入新的推荐结果 insert_query = """ INSERT INTO recommendations (user_id, music_id, recommendation_score, reason) VALUES (%s, %s, %s, %s) """ for i, rec in enumerate(recommendations): score = 1.0 - (i * 0.1) # 根据排名计算分数 reason = f"基于您的听歌喜好和相似用户推荐" cursor.execute(insert_query, (user_id, rec['id'], score, reason)) connection.commit() except Error as e: print(f"保存推荐结果错误: {e}") connection.rollback() finally: cursor.close() if __name__ == '__main__': app.run(debug=True)