简介这份资源是一套面向高校学生与Python初学者的电影推荐视频网站完整项目以协同过滤算法为核心适合用作毕业设计、期末大作业或课程设计的高分参考。项目包含源代码、配套论文与SQL数据库文件代码附有详细注释新手也能读懂简单部署后即可运行使用。压缩包共687个文件约13.19MB其中38个py文件承载推荐算法与后端逻辑33个vue与162个js文件构成前端交互界面另有51个css、30个png与26个jpg等样式和图片素材2个sql文件提供数据库结构并附带bat一键安装与运行脚本目录组织清晰。目前已有323人学习下载。读者可获得完整的协同过滤推荐实现思路、可运行的网站源码、论文撰写参考以及数据库建表脚本便于快速搭建环境、理解算法流程并完成二次开发是兼顾学习与实战的综合性项目资料。1. 从零搭一套电影推荐视频网站协同过滤到底解决了什么问题你打开一个电影网站首页推荐位永远摆着那几部运营手动挑的片子用户翻两页就关掉了。这个场景几乎每个做过内容站的人都遇到过片库有几千部但用户找不到自己想看的停留时长上不去回访率也低。Python 基于协同过滤算法实现的电影推荐视频网站要解决的就是这件事——让系统根据用户的历史评分行为自动算出他可能喜欢的电影把推荐位从「运营拍脑袋」变成「算法算出来」。这套方案通常包含三块一份可跑的 Python 推荐算法代码、一套视频网站的前后端源码、一份建库用的 sql 文件外加一篇把算法和系统讲清楚的论文。适合谁正在做毕业设计的学生、想给自己小站加推荐功能的独立开发者、以及想搞明白协同过滤从公式到上线全链路的初级工程师。下面我按「算法怎么算 → 数据怎么存 → 网站怎么搭 → 坑在哪」的顺序把这条链路拆开讲。2. 协同过滤的两条路线UserCF 和 ItemCF 怎么选、怎么算2.1 用户相似度和物品相似度差在一个矩阵转置协同过滤的核心思想很朴素找到和你口味相似的人把他们喜欢而你没看过的电影推给你这叫 UserCF或者找到和你看过这部电影相似的其他电影推给你这叫 ItemCF。两者数学上都是算相似度区别在于把「用户-物品评分矩阵」拿谁来算。假设评分矩阵 R 是 m 个用户 × n 部电影UserCF 算的是行与行之间的相似度ItemCF 算的是列与列之间的相似度。代码上就是一次转置的差别import numpy as np from sklearn.metrics.pairwise import cosine_similarity # R: 用户-电影评分矩阵行用户列电影0 表示未评分 R np.array([ [5, 3, 0, 1], [4, 0, 0, 1], [1, 1, 0, 5], [1, 0, 0, 4], [0, 1, 5, 4], ]) # UserCF用户之间的相似度直接对行算 user_sim cosine_similarity(R) # 形状 (5, 5) # ItemCF电影之间的相似度转置后对列算 item_sim cosine_similarity(R.T) # 形状 (4, 4) print(用户相似度矩阵:\n, np.round(user_sim, 2)) print(电影相似度矩阵:\n, np.round(item_sim, 2))这段代码里cosine_similarity默认按行计算所以算 ItemCF 时必须先R.T把矩阵转过来。参数上R里的 0 代表「没评过分」不是「评了 0 分」这个区分后面会反复提到处理错了整个推荐结果就废了。相似度取值范围 0 到 1越接近 1 越相似。2.2 预测评分的公式和 TopN 推荐的生成算出相似度只是第一步真正要的是「预测某个用户对某部没看过的电影打几分」。UserCF 的预测公式是用和目标用户最相似的 K 个用户的评分按相似度加权平均。def predict_user_cf(R, user_sim, user_id, item_id, k2): 基于 UserCF 预测 user_id 对 item_id 的评分 # 找出对 item_id 评过分、且不是自己的用户 rated_users np.where(R[:, item_id] 0)[0] rated_users rated_users[rated_users ! user_id] if len(rated_users) 0: return 0 # 没人评过无法预测 # 按相似度取前 k 个邻居 sims user_sim[user_id, rated_users] top_k_idx np.argsort(sims)[::-1][:k] neighbors rated_users[top_k_idx] neighbor_sims sims[top_k_idx] # 相似度加权平均 if neighbor_sims.sum() 0: return 0 pred np.sum(neighbor_sims * R[neighbors, item_id]) / neighbor_sims.sum() return pred # 预测用户 0 对电影 2 的评分 score predict_user_cf(R, user_sim, user_id0, item_id2, k2) print(f用户0对电影2的预测评分: {score:.2f})k2是邻居数量这是协同过滤最关键的参数之一k 太小推荐容易被个别极端用户带偏k 太大又会把不相似的人也算进来推荐变得平庸。实践中 k 一般取 20 到 50小数据集上取 5 到 10 也能跑。neighbor_sims.sum() 0这个判断是防止除零真实数据里经常出现某部电影只有一个人评过、相似度又恰好为 0 的情况。生成 TopN 推荐就是把用户所有没看过的电影都预测一遍按分数排序取前 N 个。这一步在网站里通常放在后端接口里用户一登录就调一次。2.3 UserCF 和 ItemCF 的选型看你的用户多还是物品多选哪个不是拍脑袋看两个指标用户数和物品数的比例以及物品的更新频率。维度UserCFItemCF适合场景用户少、物品多、时效性强用户多、物品少、物品稳定典型例子新闻推荐、短视频电影、电商、图书相似度矩阵规模用户数 × 用户数物品数 × 物品数可解释性差「和你相似的人喜欢」好「和你看过的相似」实时更新新用户加入要重算新物品加入只影响一列电影网站我一般选 ItemCF。原因很直接电影数量相对稳定用户数量会持续增长ItemCF 的相似度矩阵规模可控而且「因为你看过《教父》所以推荐《美国往事》」这种解释用户能看懂点击率明显比 UserCF 的「和你相似的人还看了」高。热搜里常出现的「python 数据分析与可视化」在这套系统里也有用武之地——把评分分布、相似度热力图用 matplotlib 画出来论文里的实验章节直接就有了。3. 数据层落地sql 文件怎么建、评分表怎么设计3.1 建库建表四张核心表撑起整个推荐系统一份能用的 sql 文件核心是四张表用户表、电影表、评分表、推荐结果表。评分表是整个协同过滤的数据来源设计得好不好直接决定算法能不能跑。-- 用户表 CREATE TABLE user ( user_id INT PRIMARY KEY AUTO_INCREMENT, username VARCHAR(50) NOT NULL UNIQUE, password VARCHAR(128) NOT NULL, create_time DATETIME DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; -- 电影表 CREATE TABLE movie ( movie_id INT PRIMARY KEY AUTO_INCREMENT, title VARCHAR(200) NOT NULL, director VARCHAR(100), genre VARCHAR(100), cover_url VARCHAR(255), video_url VARCHAR(255), avg_score DECIMAL(3,1) DEFAULT 0.0 ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; -- 评分表协同过滤的唯一数据源 CREATE TABLE rating ( id BIGINT PRIMARY KEY AUTO_INCREMENT, user_id INT NOT NULL, movie_id INT NOT NULL, score TINYINT NOT NULL COMMENT 1-5 分, rate_time DATETIME DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uk_user_movie (user_id, movie_id), KEY idx_movie (movie_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; -- 推荐结果表离线算好在线直接查 CREATE TABLE recommend ( id BIGINT PRIMARY KEY AUTO_INCREMENT, user_id INT NOT NULL, movie_id INT NOT NULL, pred_score DECIMAL(4,2), gen_time DATETIME DEFAULT CURRENT_TIMESTAMP, KEY idx_user (user_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;几个设计点值得说清楚。rating表上的UNIQUE KEY uk_user_movie保证一个用户对一部电影只能有一条评分避免重复数据污染相似度计算。idx_movie索引是为了按电影聚合评分时快一点。recommend表是典型的空间换时间——协同过滤全量算一次可能要几秒到几十秒不可能每次请求都算所以离线跑完存进表里在线接口直接SELECT就行。3.2 从 sql 文件导入到 Python 读数据拿到 sql 文件后导入和读取是两件事。导入用命令行读取用 Python 连接库。# 建库并导入 sql 文件 mysql -u root -p -e CREATE DATABASE movie_rec DEFAULT CHARSET utf8mb4; mysql -u root -p movie_rec movie_rec.sql # 验证导入结果 mysql -u root -p movie_rec -e SELECT COUNT(*) FROM rating; SELECT COUNT(*) FROM movie;import pymysql import pandas as pd import numpy as np conn pymysql.connect( hostlocalhost, userroot, passwordyour_password, databasemovie_rec, charsetutf8mb4 ) # 读出评分数据直接构造成矩阵 sql SELECT user_id, movie_id, score FROM rating df pd.read_sql(sql, conn) print(f评分记录数: {len(df)}, 用户数: {df.user_id.nunique()}, 电影数: {df.movie_id.nunique()}) # 透视成用户-电影矩阵缺失值填 0 R df.pivot_table(indexuser_id, columnsmovie_id, valuesscore, fill_value0).values print(评分矩阵形状:, R.shape) conn.close()pivot_table这一步是把长表转成宽表fill_value0把没评分的填成 0。这里有个容易翻车的点如果用户 ID 或电影 ID 不连续比如删过数据透视出来的矩阵列顺序和原始 ID 对不上后面推荐出来的电影 ID 会错位。稳妥做法是保留pivot_table的columns索引用columns.get_loc(movie_id)反查位置而不是直接用电影 ID 当数组下标。提示导入 sql 文件时如果报字符集错误先确认文件本身是 utf8mb4 编码再检查连接串里的charset参数两个地方不一致是最常见的原因。4. 网站层实现Flask 接口 推荐结果怎么串起来4.1 后端接口三个路由撑起推荐闭环网站后端不需要多复杂Flask 起三个接口就够跑通登录、拿推荐列表、提交评分。推荐列表接口从recommend表读离线算好的结果评分接口写回rating表并触发重新计算。from flask import Flask, request, jsonify import pymysql app Flask(__name__) def get_conn(): return pymysql.connect(hostlocalhost, userroot, passwordyour_password, databasemovie_rec, charsetutf8mb4, cursorclasspymysql.cursors.DictCursor) app.route(/api/recommend/int:user_id) def recommend(user_id): 返回该用户的 TopN 推荐电影 conn get_conn() with conn.cursor() as cur: cur.execute( SELECT m.movie_id, m.title, m.cover_url, r.pred_score FROM recommend r JOIN movie m ON r.movie_id m.movie_id WHERE r.user_id %s ORDER BY r.pred_score DESC LIMIT 10 , (user_id,)) rows cur.fetchall() conn.close() return jsonify({code: 0, data: rows}) app.route(/api/rate, methods[POST]) def rate(): 提交评分写入 rating 表 data request.get_json() user_id, movie_id, score data[user_id], data[movie_id], data[score] if not (1 score 5): return jsonify({code: 1, msg: 评分必须在 1-5 之间}) conn get_conn() with conn.cursor() as cur: cur.execute( INSERT INTO rating (user_id, movie_id, score) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE score VALUES(score) , (user_id, movie_id, score)) conn.commit() conn.close() return jsonify({code: 0, msg: ok}) if __name__ __main__: app.run(debugTrue, port5000)ON DUPLICATE KEY UPDATE配合前面建的唯一索引实现「评过就更新、没评过就插入」省掉一次查询。pred_score直接从推荐表读接口响应时间基本在毫秒级。评分接口写完数据后推荐结果不会立刻变——需要离线任务重新跑一遍这是这套架构的固有延迟后面避坑章节会讲怎么缓解。4.2 离线计算任务把算法结果写回数据库离线任务就是定时把第 2 章的算法跑一遍结果写进recommend表。用schedule或者系统的 crontab 都行。import schedule import time import numpy as np import pymysql from sklearn.metrics.pairwise import cosine_similarity def run_recommend_job(): conn pymysql.connect(hostlocalhost, userroot, passwordyour_password, databasemovie_rec, charsetutf8mb4) # 1. 读评分矩阵 import pandas as pd df pd.read_sql(SELECT user_id, movie_id, score FROM rating, conn) if len(df) 10: print(评分数据太少跳过本次计算) conn.close() return R df.pivot_table(indexuser_id, columnsmovie_id, valuesscore, fill_value0) user_ids R.index.tolist() movie_ids R.columns.tolist() mat R.values # 2. ItemCF 相似度 item_sim cosine_similarity(mat.T) # 3. 为每个用户生成 TopN results [] for i, uid in enumerate(user_ids): scores [] for j, mid in enumerate(movie_ids): if mat[i, j] 0: continue # 已看过的跳过 # 用该用户评过分的电影加权算目标电影得分 rated_idx np.where(mat[i] 0)[0] if len(rated_idx) 0: continue sims item_sim[j, rated_idx] pred np.sum(sims * mat[i, rated_idx]) / (sims.sum() 1e-8) scores.append((mid, pred)) scores.sort(keylambda x: x[1], reverseTrue) for mid, pred in scores[:10]: results.append((uid, mid, round(float(pred), 2))) # 4. 写回数据库 with conn.cursor() as cur: cur.execute(DELETE FROM recommend) cur.executemany( INSERT INTO recommend (user_id, movie_id, pred_score) VALUES (%s, %s, %s), results) conn.commit() conn.close() print(f推荐计算完成写入 {len(results)} 条) # 每小时跑一次 schedule.every(1).hours.do(run_recommend_job) run_recommend_job() # 启动时先跑一次 while True: schedule.run_pending() time.sleep(60)1e-8是防止相似度和为 0 时除零。DELETE FROM recommend再批量插入比逐条更新简单数据量不大时完全够用。executemany批量插入比循环单条插入快一个数量级几千条推荐结果一两秒就写完。这套离线任务的计算复杂度是用户数 × 电影数 × 平均评分数数据量上万后单机跑会变慢优化方向是只算活跃用户、或者用矩阵运算替代双重循环。4.3 前端展示推荐位和评分组件的最小实现前端不用框架也能跑一个 HTML 页面加 fetch 请求就够。推荐位渲染成卡片列表每张卡片带一个五星评分组件。// 拉取推荐列表并渲染 async function loadRecommend(userId) { const res await fetch(/api/recommend/${userId}); const json await res.json(); const box document.getElementById(recommend-list); box.innerHTML json.data.map(m div classmovie-card img src${m.cover_url} alt${m.title} h3${m.title}/h3 p预测评分: ${m.pred_score}/p div classstars>def evaluate_precision_recall(R_train, R_test, recommend_fn, k10): 计算 PrecisionK 和 RecallK precisions, recalls [], [] n_users R_train.shape[0] for uid in range(n_users): # 测试集里该用户评过 4 分以上的电影作为「真正喜欢」 true_items set(np.where(R_test[uid] 4)[0]) if not true_items: continue # 推荐列表 rec_items set(recommend_fn(R_train, uid, k)) hit true_items rec_items precisions.append(len(hit) / k) recalls.append(len(hit) / len(true_items)) return np.mean(precisions), np.mean(recalls) # 假设已有 recommend_fn 返回 TopK 电影索引 p, r evaluate_precision_recall(R_train, R_test, my_recommend_fn, k10) print(fPrecision10: {p:.3f}, Recall10: {r:.3f})R_test[uid] 4这个阈值是定义「用户真正喜欢」的标准取 4 分还是 3 分要看你的评分分布一般取中位数以上。Precision 高说明推得准Recall 高说明推得全两者通常此消彼长调 k 值就是在两者之间找平衡。覆盖率是推荐出的不同电影数除以总电影数太低说明系统只推那几部热门多样性差。离线指标好看不等于线上效果好最终还得靠 A/B 测试。最小做法是把用户随机分两组一组看协同过滤推荐一组看热门推荐跑一周对比点击率和评分转化率。我自己的习惯是离线指标只用来筛掉明显不行的方案真正决定上不上线看 A/B 数据。有个血泪教训曾经离线 Precision 提升了 8%上线后点击率反而降了后来发现是推荐太集中在少数几部电影上用户刷两下就腻了。从那以后我每次都会同时看覆盖率和推荐列表的多样性不再只盯准确率。希望这些能帮你在自己的电影推荐项目里少走点弯路。本文还有配套的精品资源点击获取