基于Django的新闻推荐系统设计与实现
一、项目概述
1.1 项目背景
随着互联网信息爆炸式增长，用户在面对海量新闻内容时难以快速找到感兴趣的信息。传统的新闻门户网站采用"一刀切"的展示方式，无法满足用户个性化需求。基于此，本项目旨在开发一个智能新闻推荐系统，通过分析用户行为数据，为用户提供个性化的新闻推荐服务。
1.2 项目目标
- 实现用户注册、登录、个人信息管理功能
- 构建新闻内容管理系统，支持新闻分类、标签管理
- 设计并实现多种推荐算法（协同过滤、内容推荐、热门推荐）
- 开发用户行为追踪系统，记录用户浏览、点击、收藏等行为
- 实现实时推荐和离线推荐相结合的混合推荐机制
- 提供推荐结果的可解释性展示
二、系统架构设计
2.1 整体架构
┌─────────────────────────────────────────────────┐
│ 前端展示层 │
│ (Vue.js + Element UI) │
└─────────────────┬───────────────┬───────────────┘
│ │
┌─────────────────▼──────┐ ┌──────▼───────────────┐
│ Django Web层 │ │ 推荐算法服务层 │
│ (视图、路由、模板) │ │ (协同过滤、内容推荐) │
└───────────────┬────────┘ └──────────┬───────────┘
│ │
┌───────────────▼──────────────────────▼───────────┐
│ 数据访问层 │
│ (ORM、Redis缓存、Elasticsearch) │
└──────────────────────┬───────────────────────────┘
│
┌──────────────────────▼───────────────────────────┐
│ 数据存储层 │
│ (MySQL、Redis、MongoDB) │
└──────────────────────────────────────────────────┘2.2 技术栈选择
- 后端框架: Django 3.2 + Django REST Framework
- 前端框架: Vue.js 3 + Element Plus
- 数据库:
- MySQL: 存储用户信息、新闻元数据
- Redis: 缓存热门新闻、用户会话、实时推荐结果
- MongoDB: 存储用户行为日志、推荐模型数据
- 搜索引擎: Elasticsearch 7.x (新闻全文检索)
- 消息队列: Celery + Redis (异步任务处理)
- 推荐算法: Surprise库 (协同过滤)、Scikit-learn (内容推荐)
三、数据库设计
3.1 核心数据表设计
# models.py 核心模型定义
from django.db import models
from django.contrib.auth.models import AbstractUser
class User(AbstractUser):
"""用户模型扩展"""
GENDER_CHOICES = (
('M', '男'),
('F', '女'),
('U', '未知'),
)
nickname = models.CharField('昵称', max_length=50, blank=True)
avatar = models.ImageField('头像', upload_to='avatars/', blank=True)
gender = models.CharField('性别', max_length=1, choices=GENDER_CHOICES, default='U')
birthday = models.DateField('生日', null=True, blank=True)
phone = models.CharField('手机号', max_length=11, blank=True)
preferences = models.JSONField('偏好设置', default=dict) # 存储用户偏好
class Meta:
db_table = 'user'
verbose_name = '用户'
verbose_name_plural = verbose_name
class Category(models.Model):
"""新闻分类"""
name = models.CharField('分类名称', max_length=50, unique=True)
slug = models.SlugField('URL标识', max_length=50, unique=True)
description = models.TextField('描述', blank=True)
order = models.IntegerField('排序', default=0)
is_active = models.BooleanField('是否激活', default=True)
class Meta:
db_table = 'category'
verbose_name = '新闻分类'
verbose_name_plural = verbose_name
class Tag(models.Model):
"""新闻标签"""
name = models.CharField('标签名称', max_length=50, unique=True)
slug = models.SlugField('URL标识', max_length=50, unique=True)
description = models.TextField('描述', blank=True)
class Meta:
db_table = 'tag'
verbose_name = '标签'
verbose_name_plural = verbose_name
class News(models.Model):
"""新闻文章"""
STATUS_CHOICES = (
('draft', '草稿'),
('published', '已发布'),
('archived', '已归档'),
)
title = models.CharField('标题', max_length=200)
content = models.TextField('内容')
summary = models.TextField('摘要', blank=True)
cover_image = models.ImageField('封面图', upload_to='news/covers/', blank=True)
author = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, verbose_name='作者')
category = models.ForeignKey(Category, on_delete=models.SET_NULL, null=True, verbose_name='分类')
tags = models.ManyToManyField(Tag, blank=True, verbose_name='标签')
source = models.CharField('来源', max_length=100, blank=True)
status = models.CharField('状态', max_length=20, choices=STATUS_CHOICES, default='draft')
is_top = models.BooleanField('置顶', default=False)
view_count = models.PositiveIntegerField('浏览数', default=0)
like_count = models.PositiveIntegerField('点赞数', default=0)
collect_count = models.PositiveIntegerField('收藏数', default=0)
comment_count = models.PositiveIntegerField('评论数', default=0)
publish_time = models.DateTimeField('发布时间', null=True, blank=True)
created_at = models.DateTimeField('创建时间', auto_now_add=True)
updated_at = models.DateTimeField('更新时间', auto_now=True)
# 用于内容推荐的向量字段
content_vector = models.TextField('内容向量', blank=True)
class Meta:
db_table = 'news'
verbose_name = '新闻'
verbose_name_plural = verbose_name
ordering = ['-publish_time', '-created_at']
indexes = [
models.Index(fields=['status', 'publish_time']),
models.Index(fields=['category', 'is_top']),
]
class UserBehavior(models.Model):
"""用户行为记录"""
BEHAVIOR_CHOICES = (
('view', '浏览'),
('click', '点击'),
('like', '点赞'),
('collect', '收藏'),
('share', '分享'),
('comment', '评论'),
)
user = models.ForeignKey(User, on_delete=models.CASCADE, verbose_name='用户')
news = models.ForeignKey(News, on_delete=models.CASCADE, verbose_name='新闻')
behavior = models.CharField('行为类型', max_length=20, choices=BEHAVIOR_CHOICES)
duration = models.PositiveIntegerField('停留时长(秒)', null=True, blank=True) # 浏览时长
score = models.FloatField('行为评分', default=1.0) # 用于推荐算法加权
created_at = models.DateTimeField('行为时间', auto_now_add=True)
class Meta:
db_table = 'user_behavior'
verbose_name = '用户行为'
verbose_name_plural = verbose_name
indexes = [
models.Index(fields=['user', 'behavior', 'created_at']),
models.Index(fields=['news', 'behavior', 'created_at']),
]
class Recommendation(models.Model):
"""推荐记录"""
TYPE_CHOICES = (
('collaborative', '协同过滤'),
('content', '内容推荐'),
('popular', '热门推荐'),
('hybrid', '混合推荐'),
)
user = models.ForeignKey(User, on_delete=models.CASCADE, verbose_name='用户')
news = models.ForeignKey(News, on_delete=models.CASCADE, verbose_name='新闻')
rec_type = models.CharField('推荐类型', max_length=20, choices=TYPE_CHOICES)
score = models.FloatField('推荐分数')
position = models.IntegerField('推荐位置') # 在推荐列表中的位置
is_shown = models.BooleanField('是否展示', default=False)
is_clicked = models.BooleanField('是否点击', default=False)
created_at = models.DateTimeField('创建时间', auto_now_add=True)
class Meta:
db_table = 'recommendation'
verbose_name = '推荐记录'
verbose_name_plural = verbose_name
indexes = [
models.Index(fields=['user', 'rec_type', 'created_at']),
]3.2 MongoDB集合设计
# 用户画像集合
user_profile = {
'user_id': ObjectId,
'interests': {
'categories': [{'category_id': 1, 'weight': 0.8}, ...],
'tags': [{'tag_id': 1, 'weight': 0.6}, ...],
'keywords': [{'keyword': '科技', 'weight': 0.9}, ...]
},
'behavior_stats': {
'total_views': 100,
'avg_duration': 45,
'preferred_time': '19:00-22:00',
'preferred_categories': ['科技', '体育']
},
'last_updated': datetime
}
# 推荐模型集合
recommendation_models = {
'model_type': 'collaborative_filtering',
'version': '1.0',
'parameters': {...},
'performance': {
'precision': 0.85,
'recall': 0.78,
'f1_score': 0.81
},
'created_at': datetime,
'is_active': True
}四、推荐算法实现
4.1 算法架构设计
# recommender/algorithms.py
import numpy as np
import pandas as pd
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
from surprise import Dataset, Reader, SVD, KNNBasic
from surprise.model_selection import cross_validate
import joblib
from django.db.models import Count, Q
from django.utils import timezone
from datetime import timedelta
import redis
import json
class BaseRecommender:
"""推荐算法基类"""
def __init__(self):
self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
def recommend(self, user_id, n=10):
"""推荐接口"""
raise NotImplementedError
def _cache_key(self, user_id, algo_type):
"""生成缓存key"""
return f"rec:{algo_type}:{user_id}"
def _get_from_cache(self, cache_key):
"""从缓存获取推荐结果"""
cached = self.redis_client.get(cache_key)
if cached:
return json.loads(cached)
return None
def _save_to_cache(self, cache_key, data, expire=3600):
"""保存推荐结果到缓存"""
self.redis_client.setex(cache_key, expire, json.dumps(data))
class PopularityRecommender(BaseRecommender):
"""热门推荐算法"""
def recommend(self, user_id=None, n=10):
cache_key = self._cache_key('global', 'popular')
cached = self._get_from_cache(cache_key)
if cached:
return cached[:n]
# 计算热门新闻（基于近期浏览、点赞、收藏加权）
from news.models import News, UserBehavior
# 获取最近7天的行为数据
week_ago = timezone.now() - timedelta(days=7)
# 计算新闻热度分数
hot_news = News.objects.filter(
status='published',
publish_time__gte=week_ago
).annotate(
hot_score=Count('userbehavior', filter=Q(
userbehavior__behavior='view',
userbehavior__created_at__gte=week_ago
)) * 0.3 +
Count('userbehavior', filter=Q(
userbehavior__behavior='like',
userbehavior__created_at__gte=week_ago
)) * 0.4 +
Count('userbehavior', filter=Q(
userbehavior__behavior='collect',
userbehavior__created_at__gte=week_ago
)) * 0.3
).order_by('-hot_score', '-publish_time')[:50]
result = [
{
'news_id': news.id,
'title': news.title,
'score': news.hot_score,
'type': 'popular'
}
for news in hot_news
]
self._save_to_cache(cache_key, result, expire=1800) # 30分钟缓存
return result[:n]
class ContentBasedRecommender(BaseRecommender):
"""基于内容的推荐算法"""
def __init__(self):
super().__init__()
self.vectorizer = TfidfVectorizer(max_features=5000, stop_words='english')
self.news_vectors = None
self.news_ids = None
def _build_feature_matrix(self):
"""构建新闻特征矩阵"""
from news.models import News
news_list = News.objects.filter(status='published').values('id', 'title', 'content', 'summary')
# 组合文本特征
texts = []
self.news_ids = []
for news in news_list:
text = f"{news['title']} {news['summary']} {news['content'][:500]}"
texts.append(text)
self.news_ids.append(news['id'])
# 计算TF-IDF向量
self.news_vectors = self.vectorizer.fit_transform(texts)
def recommend(self, user_id, n=10):
cache_key = self._cache_key(user_id, 'content')
cached = self._get_from_cache(cache_key)
if cached:
return cached[:n]
# 如果特征矩阵未构建，先构建
if self.news_vectors is None:
self._build_feature_matrix()
from news.models import UserBehavior, News
# 获取用户最近浏览的新闻
user_behaviors = UserBehavior.objects.filter(
user_id=user_id,
behavior__in=['view', 'like', 'collect']
).order_by('-created_at')[:20]
if not user_behaviors:
# 如果没有历史行为，返回空或热门推荐
return []
# 获取用户浏览新闻的特征向量
user_news_ids = [ub.news_id for ub in user_behaviors]
user_news_indices = [self.news_ids.index(nid) for nid in user_news_ids if nid in self.news_ids]
if not user_news_indices:
return []
# 计算用户偏好向量（加权平均）
user_vector = np.mean(self.news_vectors[user_news_indices].toarray(), axis=0)
# 计算所有新闻与用户偏好的相似度
similarities = cosine_similarity([user_vector], self.news_vectors.toarray())[0]
# 排除用户已经看过的新闻
viewed_news = set(user_news_ids)
# 获取相似度最高的新闻
news_scores = []
for idx, news_id in enumerate(self.news_ids):
if news_id not in viewed_news:
news_scores.append((news_id, similarities[idx]))
# 按相似度排序
news_scores.sort(key=lambda x: x[1], reverse=True)
result = []
for news_id, score in news_scores[:n]:
try:
news = News.objects.get(id=news_id)
result.append({
'news_id': news_id,
'title': news.title,
'score': float(score),
'type': 'content'
})
except News.DoesNotExist:
continue
self._save_to_cache(cache_key, result, expire=3600)
return result
class CollaborativeFilteringRecommender(BaseRecommender):
"""协同过滤推荐算法"""
def __init__(self):
super().__init__()
self.model = None
self.user_ids = None
self.news_ids = None
def _load_training_data(self):
"""加载训练数据"""
from news.models import UserBehavior
# 获取用户行为数据
behaviors = UserBehavior.objects.filter(
behavior__in=['view', 'like', 'collect']
).values('user_id', 'news_id', 'score')
# 转换为Surprise需要的格式
data = []
for behavior in behaviors:
data.append({
'user': str(behavior['user_id']),
'item': str(behavior['news_id']),
'rating': behavior['score']
})
return pd.DataFrame(data)
def train_model(self):
"""训练模型"""
df = self._load_training_data()
if len(df) < 100: # 数据量太少，不训练
return False
reader = Reader(rating_scale=(0, 5))
data = Dataset.load_from_df(df[['user', 'item', 'rating']], reader)
# 使用SVD算法
self.model = SVD(n_factors=50, n_epochs=20, lr_all=0.005, reg_all=0.02)
# 训练模型
trainset = data.build_full_trainset()
self.model.fit(trainset)
# 保存用户和物品ID映射
self.user_ids = list(df['user'].unique())
self.news_ids = list(df['item'].unique())
# 保存模型
joblib.dump(self.model, 'models/collaborative_filtering.pkl')
return True
def recommend(self, user_id, n=10):
cache_key = self._cache_key(user_id, 'collaborative')
cached = self._get_from_cache(cache_key)
if cached:
return cached[:n]
# 作者声明:本文包含人工智能生成内容。
