前言

本文将深入解析完整的智能电影推荐系统项目的后端实现，涵盖用户认证、智能推荐、AI对话、管理员后台等核心模块，适合想要了解推荐系统实战开发的开发者。如果说优秀的前端设计是为了让项目有一个美观的用户交互界面，那么其相对应的后端就是整个项目中不可或缺的灵魂。后端中的各种数据是前端界面展示的基本盘，在本智能电影推荐系统中，后端不仅存储存了前端所需要的数据，在收到前端的数据请求后，还能通过API地址通信将数据直接返回到前端，从而能够实现方便快捷的数据交换功能。

后端技术栈

技术	用途
Flask 2.0+	Web框架，处理HTTP请求和路由
SQLAlchemy	ORM工具，数据库操作
MySQL 8.0	数据持久化存储
Joblib	机器学习模型序列化与加载
scikit-learn	推荐算法实现（TF-IDF、余弦相似度）
OpenAI API (通义千问)	AI智能对话功能
Flask-CORS	处理跨域请求，支持Session
Werkzeug	密码哈希加密与验证

系统架构设计

前端(Vue 3)         API接口层(Flask)     业务逻辑层             数据存储层(MySQL/Joblib)
用户界面              路由分发                 推荐算法模块                      电影数据
Session管理            AI对话模块                          用户数据
CRUD操作                           模型文件

核心功能模块详解

用户认证模块 (/api/auth/)

设计特点：双角色登录系统（普通用户 + 管理员），Session-based认证

# 统一登录接口，支持角色区分
@app.route('/api/auth/login', methods=['POST'])
def login():
    username = request.json.get('username')
    password = request.json.get('password')
    role = request.json.get('role', 'client')  # 默认为普通用户
    
    # 根据角色查询不同表
    if role == 'admin':
        user = Admin.query.filter_by(username=username).first()
        session_key = 'admin_id'
    else:
        user = Client.query.filter_by(username=username).first()
        session_key = 'client_id'
    
    # 密码校验（哈希比对）
    if user and check_password_hash(user.password, password):
        session.clear()
        session[session_key] = user.id
        return jsonify({"msg": "登录成功", "role": role})

安全特性：

密码双重存储：password（哈希加密）用于登录验证 + password_plain（明文）供管理员查看

Session清理机制：登录前清除旧的Session，防止会话固定攻击

角色隔离：普通用户和管理员使用不同的Session key

2. 电影推荐模块 (/api/movies/)

(1) 首页个性化推荐 (/api/movies/home)

算法逻辑：

def home_data():
    is_logged_in = 'client_id' in session
    
    if is_logged_in and MODEL:  # 登录用户 + 模型可用
        # 获取用户最后观看的电影
        last_foot = Footprint.query.filter_by(
            client_id=session['client_id']
        ).order_by(Footprint.id.desc()).first()
        
        if last_foot and last_foot.movie_id in MODEL['indices']:
            # 基于内容的相似度推荐
            idx = MODEL['indices'][last_foot.movie_id]
            sim_scores = list(enumerate(MODEL['content_sim'][idx]))
            # 取最相似的12部电影（排除自身）
            sim_scores = sorted(sim_scores, key=lambda x: x[1], reverse=True)[1:13]
            recs_ids = [MODEL['movie_ids'][i] for i, score in sim_scores]
            recs_raw = Movie.query.filter(Movie.id.in_(recs_ids)).all()
        else:
            # 无历史记录时随机推荐
            recs_raw = Movie.query.order_by(func.rand()).limit(12).all()
    else:
        # 未登录用户：随机推荐
        recs_raw = Movie.query.order_by(func.rand()).limit(12).all()
    
    return jsonify({"recs": [m.to_dict() for m in recs_raw]})

(2) 详情页"猜你喜欢" (/api/movies/detail/<int:mid>)

特点：登录/未登录用户差异化推荐策略

@app.route('/api/movies/detail/<int:mid>')
def movie_detail(mid):
    movie = Movie.query.get_or_404(mid)
    movie.click_count += 1  # 热度统计
    
    # 记录用户足迹（仅登录用户）
    if 'client_id' in session:
        db.session.add(Footprint(
            client_id=session['client_id'], 
            movie_id=mid
        ))
    
    # 相似推荐逻辑
    if 'client_id' in session and MODEL and mid in MODEL['indices']:
        # 登录用户：使用模型计算的内容相似度
        idx = MODEL['indices'][mid]
        sim_scores = list(enumerate(MODEL['content_sim'][idx]))
        sim_scores = sorted(sim_scores, key=lambda x: x[1], reverse=True)[1:7]
        related_movies = []
        for i, score in sim_scores:
            rid = MODEL['movie_ids'][i]
            rm = db.session.get(Movie, rid)
            if rm: 
                related_movies.append(rm.to_dict())
    else:
        # 未登录用户：基于标签的简单推荐
        tag = movie.tags.split(',')[0] if movie.tags else ""
        rel = Movie.query.filter(
            Movie.tags.like(f'%{tag}%'), 
            Movie.id != mid
        ).limit(6).all()
        related_movies = [r.to_dict() for r in rel]
    
    result = movie.to_dict()
    result['related'] = related_movies
    return jsonify(result)

3. 数据模型设计 (models.py)

class Movie(db.Model):
    __tablename__ = 'movie'
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(255), nullable=False)
    tags = db.Column(db.String(255))  # 逗号分隔的标签
    director = db.Column(db.String(100))
    actors = db.Column(db.String(500))
    summary = db.Column(db.Text)
    score = db.Column(db.Float, default=0.0)
    click_count = db.Column(db.Integer, default=0)  # 热度统计
    release_date = db.Column(db.Date)
    
    # 级联删除：删除电影时自动删除相关评分和足迹
    ratings = db.relationship('Rating', backref='movie', 
                             cascade="all, delete-orphan")
    footprints = db.relationship('Footprint', backref='movie', 
                               cascade="all, delete-orphan")
    
    def to_dict(self):
        """标准化输出格式，供前端使用"""
        return {
            "id": self.id,
            "title": self.title,
            "tags": self.tags or "剧情",
            "director": self.director or "未知",
            "actors": self.actors or "未知",
            "summary": self.summary or "暂无简介",
            "score": round(self.score or 0, 1),
            "click_count": self.click_count or 0,
            "release_date": self.release_date.isoformat() 
                          if self.release_date else "2024-01-01"
        }

4. AI智能对话模块 (/api/ai/chat)

集成通义千问API，提供电影咨询服务：

@app.route('/api/ai/chat', methods=['POST'])
def ai_chat():
    try:
        user_msg = request.json.get('message')
        # 使用OpenAI兼容接口调用通义千问
        client = OpenAI(
            api_key="sk-026d93af62d54d078aadf6011d79b2fc",
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
        )
        
        completion = client.chat.completions.create(
            model="qwen-plus",
            messages=[
                {
                    "role": "system", 
                    "content": "你是一个电影专家，请简洁地回答用户关于电影的问题。"
                },
                {"role": "user", "content": user_msg},
            ]
        )
        
        ai_reply = completion.choices[0].message.content
        return jsonify({"reply": ai_reply})
        
    except Exception as e:
        print(f"AI Error: {e}")
        return jsonify({"reply": "抱歉，我现在的思维有点混乱，请稍后再试。"}), 500

5. 管理员后台模块

(1) 数据统计仪表盘 (/api/admin/stats)

@app.route('/api/admin/stats')
def get_admin_stats():
    # 实时统计
    movie_count = Movie.query.count()
    client_count = Client.query.count()
    total_clicks = db.session.query(func.sum(Movie.click_count)).scalar() or 0
    
    # 从模型文件中读取评估指标
    rmse = precision = "N/A"
    if os.path.exists(MODEL_PATH):
        temp_model = joblib.load(MODEL_PATH)
        if 'metrics' in temp_model:
            rmse = temp_model['metrics'].get('rmse', "N/A")
            precision = temp_model['metrics'].get('precision', "N/A")
    
    return jsonify({
        "movie_count": movie_count,
        "client_count": client_count,
        "total_clicks": int(total_clicks),
        "rmse": rmse,
        "precision": precision
    })

(2) 用户审计功能 (/api/admin/clients)

特色：可查看用户的明文密码（仅限管理员）

@app.route('/api/admin/clients')
def admin_clients():
    clients = Client.query.all()
    return jsonify([
        {
            "id": c.id,
            "username": c.username,
            "password": c.password_plain,  # 返回明文密码
            "date": c.created_at.strftime('%Y-%m-%d')
        } for c in clients
    ])

推荐功能算法实现

1. 模型训练流程 (train_evaluate_model.py)

def process_recommender():
    with app.app_context():
        # 1. 数据加载
        ratings_df = pd.read_sql("SELECT user_id, movie_id, rating FROM rating", db.engine)
        
        # 2. 训练测试分割 (80/20)
        train_df, test_df = train_test_split(ratings_df, test_size=0.2, random_state=42)
        
        # 3. 内容相似度计算（TF-IDF + 余弦相似度）
        movie_data = [{'id': m.id, 'text': f"{m.tags} {m.director} {m.actors} {m.summary}"} 
                      for m in Movie.query.all()]
        tfidf = TfidfVectorizer(stop_words='english')
        tfidf_matrix = tfidf.fit_transform([m['text'] for m in movie_data])
        content_sim = cosine_similarity(tfidf_matrix)
        
        # 4. 协同过滤相似度（ItemCF）
        user_item_matrix = train_df.pivot_table(
            index='user_id', columns='movie_id', values='rating'
        ).fillna(0)
        item_cf_sim = cosine_similarity(user_item_matrix.T)
        
        # 5. 模型评估
        movie_means = train_df.groupby('movie_id')['rating'].mean()
        global_mean = train_df['rating'].mean()
        
        y_true = test_df['rating']
        y_pred = test_df['movie_id'].map(movie_means).fillna(global_mean)
        
        rmse_val = np.sqrt(mean_squared_error(y_true, y_pred))
        precision_val = (np.abs(y_true - y_pred) <= 1.0).sum() / len(test_df)
        
        # 6. 模型打包保存
        model_pack = {
            'content_sim': content_sim,
            'item_cf_sim': item_cf_sim,
            'movie_ids': [m['id'] for m in movie_data],
            'indices': {m['id']: i for i, m in enumerate(movie_data)},
            'metrics': {
                'rmse': round(float(rmse_val), 4),
                'precision': f"{round(float(precision_val) * 100, 2)}%",
                'eval_date': datetime.now().strftime('%Y-%m-%d %H:%M')
            }
        }
        joblib.dump(model_pack, MODEL_PATH)

2. 模型文件结构

# 加载模型示例
MODEL = joblib.load(MODEL_PATH)

# MODEL 数据结构：
# {
#     'content_sim': np.ndarray,  # 内容相似度矩阵 (n_movies × n_movies)
#     'item_cf_sim': np.ndarray,  # 协同过滤相似度矩阵
#     'movie_ids': List[int],     # 电影ID列表，对应矩阵行索引
#     'indices': Dict[int, int],  # {电影ID: 矩阵索引}
#     'metrics': {
#         'rmse': float,          # 均方根误差
#         'precision': str,       # 准确率（如 "65.23%"）
#         'eval_date': str        # 评估时间
#     }
# }

数据库初始化与维护

数据初始化 (init_dataset.py)

def init_all():
    with app.app_context():
        # 重建所有表
        db.drop_all()
        db.create_all()
        
        # 创建默认管理员账号
        admin = Admin(
            username="root",
            password=generate_password_hash("admin123")
        )
        db.session.add(admin)
        
        # 导入MovieLens 100k数据
        items = pd.read_csv('u.item', sep='|', encoding='latin-1')
        for _, row in items.iterrows():
            movie = Movie(
                id=row['id'],
                title=row['title'],
                # 解析标签、导演、演员等信息...
            )
            db.session.add(movie)
        
        # 导入评分数据（10万条）
        ratings = pd.read_csv('u.data', sep='\t', names=['uid', 'mid', 'rating', 'ts'])
        # 批量插入优化...
        
        db.session.commit()

海报下载工具 (download_posters.py)

# 从GitHub仓库批量下载电影海报
csv_url = "https://raw.githubusercontent.com/babu-thomas/movielens-posters/master/movie_poster.csv"
df = pd.read_csv(csv_url, names=['id', 'url'])

for _, row in df.iterrows():
    movie_id = row['id']
    img_url = row['url']
    target_file = os.path.join(save_path, f"{movie_id}.jpg")
    
    # 下载并保存海报
    r = requests.get(img_url, timeout=10)
    if r.status_code == 200:
        with open(target_file, 'wb') as f:
            f.write(r.content)

运行与依赖

启动步骤

# 1. 初始化数据库
python init_dataset.py

# 2. 训练推荐模型
python train_evaluate_model.py

# 3. 下载电影海报
python download_posters.py

# 4. 启动Flask服务
python app.py
# 服务运行在 http://127.0.0.1:5000

关键配置项

# app.py中的关键配置
app.secret_key = 'cinema_secret_v10'  # Session加密密钥
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:123456@localhost:3306/moviedb'
MODEL_PATH = r'F:\Projects\ZhangKKKKS\Back\trained_models.pkl'  # 模型文件路径

性能优化与特色

1. 性能优化

分级推荐策略：未登录用户 → 随机推荐；登录用户 → 个性化推荐

模型预加载：启动时加载模型到内存，减少IO开销

缓存机制：频繁访问的数据可考虑加入Redis缓存

2. 安全性设计

密码双重存储：哈希验证 + 明文审计（仅管理员可见）

SQL注入防护：使用SQLAlchemy ORM，自动参数化查询

Session管理：登录前清理旧Session，防止会话劫持

3. 可扩展性

模块化设计：各功能模块独立，便于扩展

算法插拔：推荐算法可通过配置切换

API标准化：RESTful接口设计，前后端解耦

总结

本电影推荐系统后端是基于Flask的完整电影推荐系统后端，整合了用户认证、智能推荐、AI对话和管理员后台等核心功能模块。通过融合基于内容的推荐算法和协同过滤算法，系统能够为不同状态的用户提供差异化的个性化推荐服务，登录用户可享受基于观影历史的精准推荐，而未登录用户则获得基于热度和随机性的普适推荐。在技术架构上，采用前后端分离设计，结合MySQL进行数据持久化，通过Joblib序列化机器学习模型，实现了从数据处理、算法训练到服务部署的完整链路。系统还特别注重安全性和可扩展性，设计了双角色认证机制、密码双重存储策略以及模块化的代码结构。整体而言，该后端不仅展示了推荐算法在实际应用中的落地方式，也为构建更复杂的推荐系统提供了可扩展的架构基础和实现参考，具备进一步集成实时推荐、深度学习模型和业务扩展的潜力。